AWS S3 文件系统如何在数分钟内导入到自己的数据处理流程中？

点击蓝字 关注我们

• Spark
• Flink
• SeaTunnel Zeta

• 批处理 
• 精确一次性

•  列投影
• 文件格式类型
• 文本
•  CSV
•  Parquet
•  ORC
•  JSON
•  Excel
  
  

如果您使用 Spark/Flink，请确保您的 Spark/Flink 集群已经集成了 Hadoop。已测试的 Hadoop 版本是 2.x。

如果您使用 SeaTunnel Zeta，则在下载和安装 SeaTunnel Zeta 时会自动集成 Hadoop jar 文件。您可以在 ${SEATUNNEL_HOME}/lib
 目录下查看这些 jar 文件。要使用此连接器，您需要将 hadoop-aws-3.1.4.jar
 和 aws-java-sdk-bundle-1.11.271.jar
放入 ${SEATUNNEL_HOME}/lib
 目录。

• 文本
• CSV
• Parquet
• ORC
• JSON
• Excel

JSON 文件类型

{"code":  200, "data":  "get success", "success":  true}

{"code":  200, "data":  "get success", "success":  true}
{"code":  300, "data":  "get failed", "success":  false}

schema {
    fields {
        code = int
        data = string
        success = boolean
    }
}

文本或 CSV 文件类型

tyrantlucifer#26#male

delimiter = "#"
schema {
    fields {
        name = string
        age = int
        gender = string 
    }
}

Orc 文件类型

Parquet文件类型

1. 在此示例中，我们从 S3 路径 s3a://seatunnel-test/seatunnel/text
    中读取数据，文件类型为 ORC。我们使用 org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider
    进行身份验证，因此需要提供 access_key
    和 secret_key
   。文件中的所有列都将被读取并发送到接收器。

# 定义运行时环境
env {
  # 您可以在这里设置 Flink 配置
  execution.parallelism = 1
  job.mode = "BATCH"
}

source {
  S3File {
    path = "/seatunnel/text"
    fs.s3a.endpoint = "s3.cn-north-1.amazonaws.com.cn"
    fs.s3a.aws.credentials.provider = "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider"
    access_key = "xxxxxxxxxxxxxxxxx"
    secret_key = "xxxxxxxxxxxxxxxxx"
    bucket = "s3a://seatunnel-test"
    file_format_type = "orc"
  }
}

transform {
  # 如果您想要获取有关如何配置 seatunnel 和查看变换插件的完整列表的更多信息，请转至 [seatunnel.apache.org/docs/category/transform-v2](https://seatunnel.apache.org/docs/category/transform-v2)。
}

sink {
  jdbc {
    url = "jdbc:mysql://localhost:3306/test"
    driver = "com.mysql.cj.jdbc.Driver"
    user = "root"
    password = "123456"
    # 基于数据库表名自动生成 SQL 语句
    generate_sink_sql = true
    database = test
    table = test_table
  }
}

1. 使用 InstanceProfileCredentialsProvider
    进行身份验证，文件类型为 JSON，因此需要配置 schema 选项。

source {
  S3File {
    path = "/seatunnel/json"
    bucket = "s3a://seatunnel-test"
    fs.s3a.endpoint = "s3.cn-north-1.amazonaws.com.cn"
    fs.s3a.aws.credentials.provider = "com.amazonaws.auth.InstanceProfileCredentialsProvider"
    file_format_type = "json"
    schema {
      fields {
        id = int 
        name = string
      }
    }
  }
}

1. 使用 InstanceProfileCredentialsProvider
    进行身份验证，文件类型为 JSON，具有五个字段 (id
   , name
   , age
   , sex
   , type
   ), 因此需要配置 schema 选项。在此作业中，我们只需将 id
    和name
    列发送到 MySQL。

env {
  # 您可以在此处设置 Flink 配置
  execution.parallelism = 1
  job.mode = "BATCH"
}

source {
  S3File {
    path = "/seatunnel/json"
    bucket = "s3a://seatunnel-test"
    fs.s3a.endpoint = "s3.cn-north-1.amazonaws.com.cn"
    fs.s3a.aws.credentials.provider = "com.amazonaws.auth.InstanceProfileCredentialsProvider"
    file_format_type = "json"
    read_columns = ["id", "name"]
    schema {
      fields {
        id = int 
        name = string
        age = int
        sex = int
        type = string
      }
    }
  }
}

transform {
  # 如果您想要获取有关如何配置 seatunnel 和查看变换插件的完整列表的更多信息，请转至 [seatunnel.apache.org/docs/category/transform-v2](https://seatunnel.apache.org/docs/category/transform-v2)。
}

sink {
  jdbc {
    url = "jdbc:mysql://localhost:3306/test"
    driver = "com.mysql.cj.jdbc.Driver"
    user = "root"
    password = "123456"
    # 基于数据库表名自动生成 SQL 语句
    generate_sink_sql = true
    database = test
    table = test_table
  }
}

2.3.0-beta 2022-10-20

• 添加了 S3File 源连接器

下一版本

• [功能] 支持 S3A 协议(3632)
• 允许用户添加其他 hadoop-s3 参数
• 允许使用 S3A 协议
• 解耦 hadoop-aws 依赖关系
• [功能] 将 S3 AK 设置为可选项 (3688)

最佳实践

测试报告