我目前有 JSON 檔案，我可以從中將其資料轉儲到臨時視圖中。遵循 Python (PySpark) 邏輯：
departMentData = spark \
.read \
.option("multiLine", True) \
.option("mode", "PERMISSIVE") \
.json("C:\\Test\data.json") \
.createOrReplaceTempView("vw_TestView")
這個臨時視圖在陣列中包含部門的資料和該部門內的員工串列。一名員工可以隸屬于多個部門。
以下是此視圖的資料型別：
- 部門 ID：字串
- 部門名稱：字串
- 員工 ID：陣列<字串。>
vw_TestView的表資料如下
| 部門ID | 部門名稱 | 員工 ID |
|---|---|---|
| D01 | 開發 | [“U1234”、“U6789”] |
| D02 | 質量保證 | [“U1234”、“U2345”] |
另一個表Employees包含所有這些員工的詳細資訊，如下所示：
| EmpID | 員工姓名 |
|---|---|
| U1234 | 喬恩 |
| U6789 | 史密斯 |
| U2345 | 娜塔莎 |
我需要新表的最終輸出如下：
| 部門ID | 部門名稱 | 員工 ID | 員工姓名 |
|---|---|---|---|
| D01 | 開發 | [“U1234”、“U6789”] | [“喬恩”，“史密斯”] |
| D02 | 質量保證 | [“U1234”、“U2345”] | [“喬恩”，“娜塔莎”] |
如何在 Databricks SQL 或 PySPark 中執行此類連接？
uj5u.com熱心網友回復：
您可以嘗試使用以下方法將explodeEmployee Id 串列拆分為不同的行，然后再加入它們并將collect_list條目聚合到一個串列中。
使用火花SQL：
注意。 確保Employees可用作表/視圖，例如EmployeeData.createOrReplaceTempView("Employees")
WITH dept_employees AS (
SELECT
DeptId,
DeptName,
explode(EmployeeIDs)
FROM
vw_TestView
)
SELECT
d.DeptId,
d.DeptName,
collect_list(e.EmpID) as EmployeeIDs,
collect_list(e.EmpName) as EmployeeNames
FROM
dept_employees d
INNER JOIN
Employees e ON d.col=e.EmpID
GROUP BY
d.Deptid,
d.DeptName
或使用 pyspark api：
from pyspark.sql import functions as F
output_df = (
departMentData.select(
F.col("DeptId"),
F.col("DeptName"),
F.explode("EmployeeIDs")
)
.alias("d")
.join(
EmployeeData.alias("e"),
F.col("d.col")==F.col("e.EmpID"),
"inner"
)
.groupBy("d.DeptId","d.DeptName")
.agg(
F.collect_list("e.EmpID").alias("EmployeeIDs"),
F.collect_list("e.EmpName").alias("EmployeeNames")
)
)
讓我知道這是否適合您。
轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/332710.html
標籤：阿帕奇火花 火花 apache-spark-sql 数据块 三角洲湖
上一篇：僅洗掉負值的前導零