我有一個如下資料集。
------------------------------------------------------- ------------------------------------------------------------------- --------------------------------------------------------------------------------------- ------------- --------------------------------------------------------- -------------
|emp_id |sik_id |modification_date |file_name |org_path |received_date|
------------------------------------------------------- ------------------------------------------------------------------- --------------------------------------------------------------------------------------- ------------- --------------------------------------------------------- -------------
|[85627230-s387s09, 98722016-s015s05, 40022035-s008s21] |[f13c1320-5c8f3daas5cd, f13c1384-6659-4831, 4831-aaf1-5c8f3da] |[2021-04-19T11:43:32.617953Z, 2021-04-19T11:43:32.858290Z, 2021-04-19T11:43:34.027082Z]|test1.json |s3://my-bucket/test_prefix/vualt2/rcvd_dt=2022-01-24/ |2022-01-25 |
|[67dm34-4334, 8723gv6-2022, 6f7m99-2244-ki856] |[66d9-4888-aaf1, aaf1-5c8f3da1d5cd, f13c1884-66d9] |[2020-11-12T23:22:05.433107Z, 2020-11-12T20:16:51.339437Z, 2020-11-11T20:59:03.758126Z]|test2.json |s3://my-bucket/test_prefix/vualt2/rcvd_dt=2022-01-25/ |2022-01-25 |
------------------------------------------------------- ------------------------------------------------------------------- --------------------------------------------------------------------------------------- ------------- --------------------------------------------------------- -------------
其架構包含陣列和字串欄位，如下所示
>>> df.printSchema()
root
|-- emp_id: array (nullable = true)
| |-- element: string (containsNull = true)
|-- sik_id: array (nullable = true)
| |-- element: string (containsNull = true)
|-- modification_date: array (nullable = true)
| |-- element: string (containsNull = true)
|-- file_name: string (nullable = false)
|-- org_path: string (nullable = false)
|-- received_date: string (nullable = false)
我想得到類似下面的結果，其中每個 emp_id、sik_id、modification_date 都得到正確的 file_name、org_path、received_date
----------------- -------------------------- ----------------------------- ------------- --------------------------------------------------------- -------------
|emp_id |sik_id |modification_date |file_name |org_path |received_date|
----------------- -------------------------- ----------------------------- ------------- --------------------------------------------------------- -------------
|85627230-s387s09 |f13c1320-5c8f3daas5cd |2021-04-19T11:43:32.617953Z |test1.json |s3://my-bucket/test_prefix/vualt2/rcvd_dt=2022-01-24/ |2022-01-25 |
|98722016-s015s05 |f13c1384-6659-4831 |2021-04-19T11:43:32.858290Z |test1.json |s3://my-bucket/test_prefix/vualt2/rcvd_dt=2022-01-24/ |2022-01-25 |
|40022035-s008s21 |4831-aaf1-5c8f3da |2021-04-19T11:43:34.027082Z |test1.json |s3://my-bucket/test_prefix/vualt2/rcvd_dt=2022-01-24/ |2022-01-25 |
|67dm34-4334 |66d9-4888-aaf1 |2020-11-12T23:22:05.433107Z |test2.json |s3://my-bucket/test_prefix/vualt2/rcvd_dt=2022-01-25/ |2022-01-25 |
|8723gv6-2022 |aaf1-5c8f3da1d5cd |2020-11-12T20:16:51.339437Z |test2.json |s3://my-bucket/test_prefix/vualt2/rcvd_dt=2022-01-25/ |2022-01-25 |
|6f7m99-2244-ki856|f13c1884-66d9 |2020-11-11T20:59:03.758126Z |test2.json |s3://my-bucket/test_prefix/vualt2/rcvd_dt=2022-01-25/ |2022-01-25 |
----------------- -------------------------- ----------------------------- ------------- --------------------------------------------------------- -------------
我嘗試在這些欄位上使用 zip() 但看起來 zip 不適用于陣列和字串欄位。當我看到型別不匹配例外時。
有人可以幫我找到正確的解決方案。
提前致謝。
uj5u.com熱心網友回復：
結合 SQL 函式arrays_zip和inline.
df = df.selectExpr('inline(arrays_zip(emp_id, sik_id, modification_date))', 'file_name', 'org_path', 'received_date')
df.show(truncate=False)
uj5u.com熱心網友回復：
嘗試爆炸：
from pyspark.sql.functions import explode
arrayData = [(['85627230-s387s09', '98722016-s015s05', '40022035-s008s21'], 'test1.json')]
df = spark.createDataFrame(data=arrayData, schema = ['emp_id', 'file_name'])
df2 = df.select(df.file_name,explode(df.emp_id))
df2.printSchema()
df2.show()
uj5u.com熱心網友回復：
除了爆炸之外，您還可以使用array_zip -
資料準備
d = {
'emp_id':[
['85627230-s387s09', '98722016-s015s05', '40022035-s008s21'],
['67dm34-4334', '8723gv6-2022', '6f7m99-2244-ki856']
] ,
'sik_id':[
['f13c1320-5c8f3daas5cd', 'f13c1384-6659-4831', '4831-aaf1-5c8f3da'],
['66d9-4888-aaf1', 'aaf1-5c8f3da1d5cd', 'f13c1884-66d9']
],
'modification_date':[
['2021-04-19T11:43:32.617953Z', '2021-04-19T11:43:32.858290Z', '2021-04-19T11:43:34.027082Z'],
['2020-11-12T23:22:05.433107Z', '2020-11-12T20:16:51.339437Z', '2020-11-11T20:59:03.758126Z']
],
'file_name': ['test1.json','test2.json'],
'org_path': ['s3://my-bucket/test_prefix/vualt2/rcvd_dt=2022-01-24/','s3://my-bucket/test_prefix/vualt2/rcvd_dt=2022-01-25/'],
'received_date': ['2022-01-25','2022-01-25']
}
df = pd.DataFrame(d)
sparkDF = sql.createDataFrame(df)
sparkDF.printSchema()
root
|-- emp_id: array (nullable = true)
| |-- element: string (containsNull = true)
|-- sik_id: array (nullable = true)
| |-- element: string (containsNull = true)
|-- modification_date: array (nullable = true)
| |-- element: string (containsNull = true)
|-- file_name: string (nullable = true)
|-- org_path: string (nullable = true)
|-- received_date: string (nullable = true)
陣列壓縮和爆炸
sparkDF = sparkDF.withColumn("exploded_tmp", F.arrays_zip( F.col('emp_id')
,F.col('sik_id')
,F.col('modification_date')
)
)\
.withColumn("exploded", F.explode("exploded_tmp"))
sparkDF.select(
F.col('exploded.emp_id')
,F.col('exploded.sik_id')
,F.col('exploded.modification_date')
,F.col('file_name')
,F.col('org_path')
,F.col('received_date')
).show()
----------------- -------------------- -------------------- ---------- -------------------- -------------
| emp_id| sik_id| modification_date| file_name| org_path|received_date|
----------------- -------------------- -------------------- ---------- -------------------- -------------
| 85627230-s387s09|f13c1320-5c8f3daa...|2021-04-19T11:43:...|test1.json|s3://my-bucket/te...| 2022-01-25|
| 98722016-s015s05| f13c1384-6659-4831|2021-04-19T11:43:...|test1.json|s3://my-bucket/te...| 2022-01-25|
| 40022035-s008s21| 4831-aaf1-5c8f3da|2021-04-19T11:43:...|test1.json|s3://my-bucket/te...| 2022-01-25|
| 67dm34-4334| 66d9-4888-aaf1|2020-11-12T23:22:...|test2.json|s3://my-bucket/te...| 2022-01-25|
| 8723gv6-2022| aaf1-5c8f3da1d5cd|2020-11-12T20:16:...|test2.json|s3://my-bucket/te...| 2022-01-25|
|6f7m99-2244-ki856| f13c1884-66d9|2020-11-11T20:59:...|test2.json|s3://my-bucket/te...| 2022-01-25|
----------------- -------------------- -------------------- ---------- -------------------- -------------
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/437352.html
標籤：Python 阿帕奇火花 pyspark apache-spark-sql
上一篇：java.lang.VerifyError：google-adsAPI和SBT的運算元堆疊溢位
下一篇：ACM證書跨賬戶DNS驗證