从ADLS的JSON文件移除元数据并通过ADF/PySpark转Parquet

步骤1：配置源数据集 #

• 创建Azure Data Lake Storage Gen2类型的源数据集，指向ADLS中存储目标JSON文件的路径。
• 格式选择JSON，结构类型设为Array of objects，确保ADF能正确解析外层数组结构。

步骤2：搭建映射数据流 #

1. 添加源转换：选择刚才配置的JSON数据集，源选项中勾选“允许schema漂移”（若JSON结构可能变动）。
2. 添加展开（Flatten）转换：将最外层的数组（默认字段名为$）展开为单独行，把每个对象拆分为单条记录。
3. 再次添加展开转换：选中records数组字段，将内部的医生数据展开为单独行。
4. 添加选择转换：移除metadata字段，仅保留展开后的医生数据字段（如doctor_id、doctor_name等）。
5. 添加接收器（Sink）转换：配置ADLS的Parquet数据集作为输出目标，设置好存储路径，按需调整分区、压缩等参数。

步骤3：运行管道 #

将数据流添加到管道中，触发运行即可完成JSON到Parquet的转换，同时自动移除元数据。

通过Synapse Spark池或本地PySpark环境连接ADLS后，执行以下代码：

步骤1：读取ADLS中的JSON文件 #

先配置ADLS访问权限（示例用SAS令牌，按需替换为Service Principal等方式）：

# 替换为你的ADLS账号和SAS令牌
spark.conf.set("fs.azure.account.auth.type.<你的ADLS账号>.dfs.core.windows.net", "SAS")
spark.conf.set("fs.azure.sas.token.provider.type.<你的ADLS账号>.dfs.core.windows.net", "org.apache.hadoop.fs.azurebfs.sas.FixedSASTokenProvider")
spark.conf.set("fs.azure.sas.fixed.token.<你的ADLS账号>.dfs.core.windows.net", "<你的SAS令牌>")

# 读取目标JSON文件
df = spark.read.json("abfss://<容器名>@<你的ADLS账号>.dfs.core.windows.net<JSON文件路径>")

步骤2：提取records数据并移除元数据 #

from pyspark.sql.functions import explode

# 展开records数组为行，同时丢弃metadata列
processed_df = df.select(explode("records").alias("doctor_info")) \
                 .select("doctor_info.*")

步骤3：写入Parquet到ADLS #

# 写入Parquet，mode设为overwrite会覆盖已有文件，可按需改为append等
processed_df.write.mode("overwrite") \
              .parquet("abfss://<容器名>@<你的ADLS账号>.dfs.core.windows.net<输出路径>")

内容的提问来源于stack exchange，提问作者SHIVAM YADAV