如何用Seatunnel读取S3 Excel与API数据并转换合并为Excel

完整配置实现 #

以下是包含字段对齐、数据合并、Excel输出的完整SeaTunnel配置：

{
  "env": {
    "job.name": "merge_user_data_job",
    "execution.parallelism": 1,
    "job.mode": "BATCH"
  },
  "source": [
    {
      "plugin_name": "S3File",
      "bucket": "s3a://bucket-name",
      "path": "/path/to/excel/file.xlsx",
      "fs.s3a.endpoint": "s3.region.amazonaws.com",
      "fs.s3a.aws.credentials.provider": "com.amazonaws.auth.InstanceProfileCredentialsProvider",
      "file_format_type": "EXCEL",
      "skip_header_row_number": 1,
      "result_table_name": "user_data_from_excel",
      "sheet": 0,
      "schema": {
        "fields": {
          "user_id": "string",
          "user_name": "string",
          "email": "string",
          "role": "string",
          "status": "string"
        }
      }
    },
    {
      "plugin_name": "Http",
      "url": "http://api-endpoint.com/api/users/graphql",
      "method": "POST",
      "headers": {
        "Content-Type": "application/json"
      },
      "body": "{\"query\":\"query GetAllUsers($filter: UserFilter) { getAllUsers(filter: $filter) { userId userName email role status createdDate lastLoginDate } }\",\"variables\":{\"filter\":{\"status\":\"ACTIVE\"}}}",
      "format": "json",
      "poll_interval": 3600,
      "result_table_name": "user_data_from_api",
      "json_field": {
        "userId": "$.data.getAllUsers[*].userId",
        "userName": "$.data.getAllUsers[*].userName",
        "email": "$.data.getAllUsers[*].email",
        "role": "$.data.getAllUsers[*].role",
        "status": "$.data.getAllUsers[*].status",
        "createdDate": "$.data.getAllUsers[*].createdDate",
        "lastLoginDate": "$.data.getAllUsers[*].lastLoginDate"
      },
      "schema": {
        "fields": {
          "userId": "string",
          "userName": "string",
          "email": "string",
          "role": "string",
          "status": "string",
          "createdDate": "string",
          "lastLoginDate": "string"
        }
      }
    }
  ],
  "transform": [
    // 对齐Excel数据源字段名，与API数据源结构统一
    {
      "plugin_name": "Sql",
      "sql": "SELECT user_id as userId, user_name as userName, email, role, status, '' as createdDate, '' as lastLoginDate FROM user_data_from_excel"
    },
    // 合并两个数据源的数据
    {
      "plugin_name": "Sql",
      "sql": "SELECT * FROM user_data_from_excel UNION ALL SELECT * FROM user_data_from_api"
    }
  ],
  "sink": [
    {
      "plugin_name": "File",
      "path": "/path/to/output/merged_users.xlsx",
      "file_format_type": "EXCEL",
      "sheet_name": "merged_user_data",
      "schema": {
        "fields": {
          "userId": "string",
          "userName": "string",
          "email": "string",
          "role": "string",
          "status": "string",
          "createdDate": "string",
          "lastLoginDate": "string"
        }
      }
    }
  ]
}

关键配置说明 #

1. 字段对齐：通过第一个Sql转换插件，将Excel数据源的user_id、user_name映射为userId、userName，同时补充API数据源存在的createdDate、lastLoginDate字段（用空字符串填充），确保两个数据源的表结构完全一致，满足合并要求。
2. 数据合并：使用UNION ALL合并两个结果表，保留所有原始数据；若需要去重合并，可替换为UNION。
3. Excel输出：配置File连接器的EXCEL格式，指定输出路径和工作表名称，输出字段需与合并后的表结构严格匹配。

内容的提问来源于stack exchange，提问作者Samir Savasani