S3 File Source Connector

Support Those Engines

Spark
Flink
SeaTunnel Zeta

Key Features

• batch
• stream
• exactly-once

Read all the data in a split in a pollNext call. What splits are read will be saved in snapshot.

• column projection
• parallelism
• support user-defined split
• file format type
  • text
  • csv
  • parquet
  • orc
  • json
  • excel
  • xml

Description

Read data from aws s3 file system.

Supported DataSource Info

Dependency

If you use spark/flink, In order to use this connector, You must ensure your spark/flink cluster already integrated hadoop. The tested hadoop version is 2.x.

If you use SeaTunnel Zeta, It automatically integrated the hadoop jar when you download and install SeaTunnel Zeta. You can check the jar package under ${SEATUNNEL_HOME}/lib to confirm this.
To use this connector you need put hadoop-aws-3.1.4.jar and aws-java-sdk-bundle-1.12.692.jar in ${SEATUNNEL_HOME}/lib dir.

Data Type Mapping

Data type mapping is related to the type of file being read, We supported as the following file types:

text csv parquet orc json excel xml

JSON File Type

If you assign file type to json, you should also assign schema option to tell connector how to parse data to the row you want.

For example:

upstream data is the following:

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

You can also save multiple pieces of data in one file and split them by newline:

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

you should assign schema as the following:

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

connector will generate data as the following:

Text Or CSV File Type

If you assign file type to text csv, you can choose to specify the schema information or not.

For example, upstream data is the following:

tyrantlucifer#26#male

If you do not assign data schema connector will treat the upstream data as the following:

If you assign data schema, you should also assign the option field_delimiter too except CSV file type

you should assign schema and delimiter as the following:

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

connector will generate data as the following:

Orc File Type

If you assign file type to parquet orc, schema option not required, connector can find the schema of upstream data automatically.

Parquet File Type

If you assign file type to parquet orc, schema option not required, connector can find the schema of upstream data automatically.

Options

delimiter/field_delimiter [string]

delimiter parameter will deprecate after version 2.3.5, please use field_delimiter instead.

compress_codec [string]

The compress codec of files and the details that supported as the following shown:

• txt: lzo none
• json: lzo none
• csv: lzo none
• orc/parquet:
  automatically recognizes the compression type, no additional settings required.

encoding [string]

Only used when file_format_type is json,text,csv,xml. The encoding of the file to read. This param will be parsed by Charset.forName(encoding).

Example

1. In this example, We read data from s3 path s3a://seatunnel-test/seatunnel/text and the file type is orc in this path. We use org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider to authentication so access_key and secret_key is required. All columns in the file will be read and send to sink.

# Defining the runtime environment
env {
  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 {
  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
    # please go to https://seatunnel.apache.org/docs/category/transform-v2
}
sink {
  Console {}
}

1. Use InstanceProfileCredentialsProvider to authentication The file type in S3 is json, so need config schema option.

  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. Use InstanceProfileCredentialsProvider to authentication The file type in S3 is json and has five fields (id, name, age, sex, type), so need config schema option. In this job, we only need send id and name column to mysql.

# Defining the runtime environment
env {
  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 {
  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
    # please go to https://seatunnel.apache.org/docs/category/transform-v2
}
sink {
  Console {}
}

Changelog

2.3.0-beta 2022-10-20

• Add S3File Source Connector

Next version

• [Feature] Support S3A protocol (3632)
  • Allow user to add additional hadoop-s3 parameters
  • Allow the use of the s3a protocol
  • Decouple hadoop-aws dependencies
• [Feature]Set S3 AK to optional (3688)