29
4
CFSDN nhấn mạnh vào giá trị tạo ra nguồn mở và chúng tôi cam kết xây dựng nền tảng chia sẻ tài nguyên để mọi nhân viên CNTT có thể tìm thấy thế giới tuyệt vời của bạn tại đây.
Bài viết trên blog CFSDN này có một số phương pháp tạo DataFrame bằng pyspark được tác giả sưu tầm và biên soạn. Nếu bạn quan tâm đến bài viết này thì nhớ like nhé.
Để thuận tiện cho việc vận hành, khi sử dụng pyspark, chúng ta thường chuyển đổi dữ liệu sang dạng DataFrame để hoàn thành các thao tác dọn dẹp và phân tích.
Trong bài viết trước về các thao tác cơ bản của pyspark đã đề cập rằng RDD cũng là một đối tượng dữ liệu phân tán cho các thao tác trong spark.
Dưới đây là cái nhìn ngắn gọn về các loại RDD và DataFrame.
|
1
2
|
in
(
kiểu
(ngày)
#
in
(
kiểu
(trích dẫn))
#
|
Sau khi xem qua các định nghĩa về mã nguồn, chúng ta có thể thấy giữa chúng không có mối quan hệ kế thừa nào.
|
1
2
3
4
5
6
7
|
lớp học
RDD (
sự vật
):
"""
Bộ dữ liệu phân tán có khả năng phục hồi (RDD), là khái niệm trừu tượng cơ bản trong Spark.
Biểu thị một tập hợp các phần tử phân vùng không thay đổi có thể được
được vận hành song song.
"""
|
|
1
2
3
4
5
6
7
|
lớp học
Khung dữ liệu(
sự vật
):
"""Một bộ sưu tập dữ liệu phân tán được nhóm thành các cột được đặt tên.
:class:`DataFrame` tương đương với một bảng quan hệ trong Spark SQL,
và có thể được tạo bằng nhiều hàm khác nhau trong :class:`SparkSession`::
...
"""
|
RDD là tập dữ liệu phân tán linh hoạt, tính trừu tượng cơ bản trong Spark. Đại diện cho một bộ sưu tập lưu trữ được phân vùng, bất biến, có thể hoạt động song song. DataFrame là một bộ sưu tập phân tán nhóm dữ liệu theo các cột tương đương với một bảng quan hệ trong Spark SQL. Điều tương tự là chúng đều được thiết kế để hỗ trợ tính toán phân tán.
Nhưng RDD chỉ là một tập hợp các phần tử, còn DataFrame được nhóm theo các cột, tương tự như các bảng của MySQL hoặc DataFrame trong gấu trúc.
Trong công việc thực tế, chúng tôi vẫn sử dụng DataFrame nhiều hơn.
Hãy thử trường hợp đầu tiên và thấy rằng nếu bạn chỉ chuyển vào bộ dữ liệu, kết quả sẽ không có tên cột. Vì vậy, chúng tôi thử tùy chọn thứ hai, chuyển cả bộ dữ liệu và tên cột.
|
1
2
3
4
5
6
7
8
|
Một
=
[(
'Alice'
,
1
)]
đầu ra
=
spark.createDataFrame(a).collect()
in
(đầu ra)
# [Hàng(_1='Alice', _2=1)]
đầu ra
=
spark.createDataFrame(a, [
'tên'
,
'tuổi'
]).sưu tầm()
in
(đầu ra)
# [Hàng(tên='Alice', tuổi=1)]
|
Ở đây coll() hiển thị từng hàng của bảng dữ liệu hoặc bạn có thể sử dụng show() để hiển thị bảng dữ liệu.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
spark.createDataFrame(a).show()
# +-----+---+
# | _1| _2|
# +-----+---+
# |Alice| 1|
# +-----+---+
spark.createDataFrame(a, [
'tên'
,
'tuổi'
]).trình diễn()
# +-----+---+
# | tên|tuổi|
# +-----+---+
# |Alice| 1|
# +-----+---+
|
|
1
2
3
4
5
|
ngày
=
[{
'tên'
:
'Alice'
,
'tuổi'
:
1
}]
đầu ra
=
spark.createDataFrame(d).collect()
in
(đầu ra)
# [Hàng(tuổi=1, tên='Alice')]
|
|
1
2
3
4
5
6
7
8
9
|
Một
=
[(
'Alice'
,
1
)]
rdd
=
sc.song song hóa(a)
đầu ra
=
spark.createDataFrame(rdd).collect()
in
(đầu ra)
đầu ra
=
spark.createDataFrame(rdd, [
"tên"
,
"tuổi"
]).sưu tầm()
in
(đầu ra)
# [Hàng(_1='Alice', _2=1)]
# [Hàng(tên='Alice', tuổi=1)]
|
|
1
2
3
4
5
6
7
8
9
10
11
|
từ
pyspark.sql
nhập khẩu
Hàng ngang
Một
=
[(
'Alice'
,
1
)]
rdd
=
sc.song song hóa(a)
Người
=
Hàng ngang(
"tên"
,
"tuổi"
)
người
=
ngày.
bản đồ
(
lambda
r: Người(
*
r))
đầu ra
=
spark.createDataFrame(người).collect()
in
(đầu ra)
# [Hàng(tên='Alice', tuổi=1)]
|
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
từ
pyspark.sql.types
nhập khẩu
*
Một
=
[(
'Alice'
,
1
)]
rdd
=
sc.song song hóa(a)
sơ đồ
=
Kiểu cấu trúc(
[
Trường cấu trúc(
"tên"
, Kiểu chuỗi(),
ĐÚNG VẬY
),
Trường cấu trúc(
"tuổi"
, Kiểu số nguyên(),
ĐÚNG VẬY
)
]
)
đầu ra
=
spark.createDataFrame(rdd, lược đồ).collect()
in
(đầu ra)
# [Hàng(tên='Alice', tuổi=1)]
|
df.toPandas() có thể chuyển đổi DataFrame pyspark thành DataFrame của gấu trúc.
|
1
2
3
4
5
6
7
8
9
10
|
df
=
spark.createDataFrame(rdd, [
'tên'
,
'tuổi'
])
in
(trích dẫn)
# DataFrame[tên: chuỗi, tuổi: bigint]
in
(
kiểu
(df.toPandas()))
#
# Truyền vào DataFrame của gấu trúc
đầu ra
=
spark.createDataFrame(df.toPandas()).collect()
in
(đầu ra)
# [Hàng(tên='Alice', tuổi=1)]
|
|
1
2
3
4
5
6
7
|
đầu ra
=
tia lửa.
phạm vi
(
1
,
7
,
2
).sưu tầm()
in
(đầu ra)
# [Hàng(id=1), Hàng(id=3), Hàng(id=5)]
đầu ra
=
tia lửa.
phạm vi
(
3
).sưu tầm()
in
(đầu ra)
# [Hàng(id=0), Hàng(id=1), Hàng(id=2)]
|
Nhận DataFrame thông qua bảng tạm thời.
|
1
2
3
4
5
|
spark.registerDataFrameAsTable(df,
"bảng1"
)
df2
=
spark.bảng(
"bảng1"
)
b
=
df.thu thập()
=
=
df2.thu thập()
in
(b)
# ĐÚNG VẬY
|
Các loại cột có thể được chỉ định trong createDataFrame và chỉ những cột đáp ứng loại dữ liệu đó mới được giữ lại. Nếu không có cột nào thỏa mãn loại dữ liệu đó thì sẽ xảy ra lỗi.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
Một
=
[(
'Alice'
,
1
)]
rdd
=
sc.song song hóa(a)
# Khi loại được chỉ định tương ứng với dữ liệu dự kiến, nó sẽ được tạo bình thường.
đầu ra
=
spark.createDataFrame(rdd,
"a: chuỗi, b: số nguyên"
).sưu tầm()
in
(đầu ra)
# [Hàng(a='Alice', b=1)]
rdd
=
ngày.
bản đồ
(
lambda
hàng: hàng[
1
])
in
(ngày)
# PythonRDD[7] tại RDD tại PythonRDD.scala:53
# Chỉ có kiểu int tương ứng và các cột khác được lọc ra.
đầu ra
=
spark.createDataFrame(rdd,
"số nguyên"
).sưu tầm()
in
(đầu ra)
# [Hàng(giá trị=1)]
# Nếu không có cột nào khớp thì sẽ xảy ra lỗi.
đầu ra
=
spark.createDataFrame(rdd,
"boolean"
).sưu tầm()
# TypeError: giá trị trường: BooleanType không thể chấp nhận đối tượng 1 trong loại
|
|
1
2
|
spark.registerDataFrameAsTable(df,
"bảng1"
)
spark.dropTempTable(
"bảng1"
)
|
|
1
2
3
4
|
in
(spark.getConf(
"spark.sql.shuffle.partitions"
))
#200
in
(spark.getConf(
"spark.sql.shuffle.partitions"
, TRONG
"10"
))
# 10
in
(spark. setConf(
"spark.sql.shuffle.partitions"
, TRONG
"50"
))
# Không có
in
(spark.getConf(
"spark.sql.shuffle.partitions"
, TRONG
"10"
))
# 50
|
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
spark.registerFunction(
"chuỗiChuỗiChiều Dài"
,
lambda
x:
chỉ một
(x))
đầu ra
=
spark.sql(
"CHỌN stringLengthString('test')"
).sưu tầm()
in
(đầu ra)
# [Hàng(stringLengthString(kiểm tra)='4')]
spark.registerFunction(
"chuỗiChuỗiChiều Dài"
,
lambda
x:
chỉ một
(x), Kiểu số nguyên())
đầu ra
=
spark.sql(
"CHỌN stringLengthString('test')"
).sưu tầm()
in
(đầu ra)
# [Hàng(stringLengthString(kiểm tra)=4)]
spark.udf.register(
"chuỗiLengthInt"
,
lambda
x:
chỉ một
(x), Kiểu số nguyên())
đầu ra
=
spark.sql(
"CHỌN stringLengthInt('kiểm tra')"
).sưu tầm()
in
(đầu ra)
# [Hàng(stringLengthInt(kiểm tra)=4)]
|
Tất cả các tên bảng và đối tượng tạm thời có thể được xem.
|
1
2
3
4
5
6
7
8
9
10
|
spark.registerDataFrameAsTable(df,
"bảng1"
)
in
(spark.tableNames())
# ['bảng1']
in
(spark.tables())
# DataFrame[cơ sở dữ liệu: chuỗi, tableName: chuỗi, isTemporary: boolean]
in
(
"bảng1"
TRONG
spark. tên bảng())
# ĐÚNG VẬY
in
(
"bảng1"
TRONG
spark. tên bảng(
"mặc định"
))
# ĐÚNG VẬY
spark.registerDataFrameAsTable(df,
"bảng1"
)
df2
=
spark.tables()
df2.
lọc
(
"tên bảng = 'bảng1'"
).Đầu tiên()
in
(df2)
# DataFrame[cơ sở dữ liệu: chuỗi, tableName: chuỗi, isTemporary: boolean]
|
Tiền đề là bạn cần tải xuống gói jar. Trình kết nối Mysql-java.jar.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
từ
pyspark
nhập khẩu
SparkContext
từ
pyspark.sql
nhập khẩu
Ngữ cảnh SQL
nhập khẩu
pyspark.sql.functions như F
sc
=
SparkContext(
"địa phương"
, tên ứng dụng
=
"kiểm tra mysql"
)
sqlBối cảnh
=
Ngữ cảnh SQL(sc)
df
=
sqlContext. đọc.
định dạng
(
"jdbc"
).tùy chọn(
địa chỉ
=
"jdbc:mysql://localhost:3306/mydata?user=root&password=mysql&"
"useUnicode=true&characterEncoding=utf-8&useJDBCCompliantTimezoneShift=true&"
"useLegacyDatetimeCode=false&serverTimezone=UTC "
, bảng cơ sở dữ liệu
=
"dữ liệu chi tiết"
).trọng tải()
df.hiển thị(n
=
5
)
sc.dừng()
|
Thẩm quyền giải quyết.
Sự khác biệt giữa RDD và DataFrame tạo ra bản dịch tài liệu chính thức pyspark.sql.SQLContext.
Đến đây là kết thúc bài viết về một số phương pháp tạo DataFrame bằng pyspark. Để biết thêm nội dung liên quan đến tạo DataFrame bằng pyspark, vui lòng tìm kiếm các bài viết trước của tôi hoặc tiếp tục duyệt qua các bài viết liên quan bên dưới. Mong các bạn sẽ ủng hộ tôi trong thời gian tới! .
Liên kết gốc: https://blog.csdn.net/weixin_39198406/article/details/104916715.
Cuối cùng, bài viết này về một số phương pháp tạo DataFrame bằng pyspark kết thúc tại đây. Nếu bạn muốn biết thêm về một số phương pháp tạo DataFrame bằng pyspark, vui lòng tìm kiếm bài viết CFSDN hoặc tiếp tục duyệt qua các bài viết liên quan. tương lai blog của tôi! .
29
4
0
Tôi có một cột gồm các số tháng trong khung dữ liệu và muốn thay đổi nó thành tên tháng, vì vậy tôi đã sử dụng cột này: df['monthName'] = df['monthNumber'].apply(lambda x: Calendar.mont
Có chức năng input() nào trong Pyspark để tôi có thể nhận thông tin đầu vào từ bảng điều khiển không. Nếu có, xin vui lòng giải thích. Cách viết đoạn mã sau trong PySpark: folder_change = input("
Chúng tôi đang xây dựng khung nhập dữ liệu trong pyspark và muốn biết cách tốt nhất để xử lý các ngoại lệ về loại dữ liệu là gì. Về cơ bản, chúng tôi muốn có một bảng từ chối ghi lại tất cả dữ liệu chưa được xác nhận với lược đồ. stringDf = sparkSession.cr
Tôi đang phát triển truy vấn sql trên khung dữ liệu tia lửa dựa trên tập hợp tệp ORC. Chương trình như thế này: from pyspark.sql import SparkSession spark_session = Spa
Tôi có khung dữ liệu Pyspark (khung dữ liệu gốc) với dữ liệu sau (tất cả các cột có kiểu dữ liệu chuỗi): id Value 1 103 2
Tôi có một máy chủ được định cấu hình với Redis và Maven, sau đó tôi thực thi sparkSession spark = pyspark .sql .SparkSession .builder .master('loca)
Từ một số thử nghiệm ngắn gọn, có vẻ như chức năng xóa cột của các khung dữ liệu pyspark không phân biệt chữ hoa chữ thường, ví dụ: từ pyspark.sql nhập SparkSession từ pyspark.sql.funct
Tôi có: +---+-------+-------+ | id var1| -+ | a|[1,2,3]|[1,2,3]|
Từ một số thử nghiệm ngắn gọn, có vẻ như chức năng xóa cột của các khung dữ liệu pyspark không phân biệt chữ hoa chữ thường, ví dụ: từ pyspark.sql nhập SparkSession từ pyspark.sql.funct
Tôi có một pyspark DF với nhiều cột số và với mỗi cột tôi muốn tính thứ hạng thập phân hoặc thứ hạng lượng tử khác của hàng dựa trên từng biến. Điều này thật dễ dàng với Pandas vì chúng ta có thể sử dụng hàm qcut để tạo cột mới cho mỗi biến, như
Tôi có đoạn mã sau để hồi quy tuyến tính bằng gói pyspark.ml. Tuy nhiên, khi mô hình phù hợp, tôi nhận được thông báo lỗi này ở dòng cuối cùng: IllegalArgumentException: u'requirement failed
Tôi có một tệp phẳng (ống) được phân cách bằng | không có ký tự trích dẫn. Dữ liệu mẫu có dạng như sau: SOME_NUMBER|SOME_MULTILINE_STRING|SOME_STRING 23|văn bản nhiều dòng
Cho mẫu sau: root |-- first_name: string |-- Last_name: string |-- Degrees: array |-- element: struc
Tôi có một khung dữ liệu pyspark như sau (đây chỉ là một ví dụ đơn giản, khung dữ liệu thực tế của tôi có hàng trăm cột): col1,col2,...,col_with_fix_header 1,2,....., 3 4,5,.
Tôi có một khung dữ liệu +------+----------------------+---------------- - +---- | id|
Tôi bắt đầu với mảng Spark "df_spark": from pyspark.sql import SparkSession import pandas as pd import numpy as np
Làm cách nào để xóa giá trị hàng trong Pyspark dựa trên số hàng/giá trị chỉ số hàng? Tôi mới sử dụng Pyspark (và mã hóa) - Tôi đã thử mã hóa thứ gì đó nhưng nó không hoạt động. Câu trả lời hay nhất Bạn không thể xóa các cột cụ thể, nhưng bạn có thể sử dụng bộ lọc hoặc bí danh của nó
Tôi có một vòng lặp tạo đầu ra của nhiều bảng yếu tố và lưu trữ tên cột trong danh sách: | id f_2a | | 1,2 |0,95 |
Tôi đang cố chuyển đổi tập lệnh hql sang pyspark. Tôi đang gặp khó khăn với cách triển khai tổng trong trường hợp khi câu lệnh được tổng hợp sau mệnh đề nhóm. Ví dụ. dataframe1 = dataframe0.gro
Tôi muốn thêm giá trị dịch vụ 2 cột mới mảng giá trị thứ nhất và thứ hai nhưng tôi gặp lỗi: Tên trường phải là Chuỗi ký tự, nhưng nó là 0;
Trung tâm cá nhân
Điều phát hành
từ bỏ
Tôi là một lập trình viên xuất sắc, rất giỏi!