如何将Script活动输出以CSV/结构化格式推送至Blob/数据库？

运行Script活动后得到JSON格式输出，请问如何将这些数据转换为CSV或其他结构化格式推送至Blob存储或数据库？另外，若Script活动（单查询）输出达100万行，该如何处理？

一、JSON转结构化格式并推送至Blob/数据库 #

推送至Blob存储 #

• 数据流活动（Data Flow）
  1. 以JSON数据集作为数据源，直接读取Script活动输出的JSON数据；
  2. 通过「选择」「派生列」等组件完成数据清洗与规整；
  3. 配置CSV/Parquet数据集作为输出sink，直接写入Blob存储，无需手动转换格式。
• 复制活动（Copy Activity）
  1. 源数据集选择JSON，指向Script活动的输出路径；
  2. 目标数据集选择CSV/Parquet等结构化格式，配置Blob存储的容器与路径；
  3. 复制活动会自动完成JSON到结构化格式的映射转换，全程可视化配置，无需代码。
• 自定义脚本转换
  若使用Python脚本活动，可借助pandas库完成转换并上传，示例代码：

  import pandas as pd
  from azure.storage.blob import BlobServiceClient
  
  # 读取JSON输出
  df = pd.read_json("./script-output.json")
  # 转换为CSV
  csv_content = df.to_csv(index=False)
  # 上传至Blob
  blob_client = BlobServiceClient.from_connection_string("YOUR_BLOB_CONN_STR").get_blob_client(container="data-container", blob="output.csv")
  blob_client.upload_blob(csv_content, overwrite=True)
  

推送至数据库 #

• 复制活动（Copy Activity）
  1. 源选JSON数据集，目标选对应数据库的数据集（如SQL Server、MySQL）；
  2. 配置字段映射规则，复制活动会自动将JSON字段映射到数据库表列，直接完成写入。
• 存储过程活动
  1. 将Script输出的JSON数据作为参数传入存储过程；
  2. 利用数据库内置函数解析JSON并插入表，以SQL Server为例：

  CREATE PROCEDURE InsertDataFromJSON
      @json_input NVARCHAR(MAX)
  AS
  BEGIN
      INSERT INTO TargetTable (id, name, create_time)
      SELECT id, name, create_time
      FROM OPENJSON(@json_input)
      WITH (
          id INT '$.id',
          name VARCHAR(50) '$.name',
          create_time DATETIME '$.create_time'
      )
  END
  

• 数据流活动
  同Blob处理逻辑，读取JSON数据后清洗，直接写入数据库sink，支持字段映射与数据校验。

二、百万行输出数据的处理方案 #

• 分页查询拆分输出
  在Script活动的查询语句中加入分页逻辑，比如SQL用OFFSET ... FETCH NEXT、MySQL用LIMIT，将百万行数据拆分为多个小批次JSON输出，再分批次转换写入，避免内存溢出。
• 数据流分区并行处理
  数据流支持按字段/范围设置分区，自动将数据拆分为多个分片并行处理，适配百万级数据量，且无需手动管理内存。
• 提升复制活动并行度
  在复制活动的设置中调高「并行复制」数量，利用多线程加速数据读写，同时确保Blob存储/数据库能承受对应并发压力。
• 优先使用高效格式
  若无需CSV，优先选择Parquet格式——它压缩率高、读写速度快，处理百万行数据时性能远优于CSV，尤其适合Blob存储场景。
• 脚本分块处理
  若用Python脚本处理，避免一次性加载全量数据，使用pandas的chunksize参数分块读取与写入：

  chunk_size = 10000
  # 分块读取JSON
  for chunk in pd.read_json("./large-output.json", chunksize=chunk_size):
      # 分块写入CSV/上传Blob
      chunk.to_csv("./output.csv", mode="a", header=False, index=False)
  

内容的提问来源于stack exchange，提问作者AzSurya Teja