Maxcompute

Maxcompute sink connector

Support Those Engines

Spark
Flink
SeaTunnel Zeta

Description

Used to write data to Maxcompute. The connector supports AccessKey (accessId/accesskey) authentication, STS-token authentication, and the default Aliyun credentials provider chain. It can append to or overwrite a target table or partition, create the target table from a template, and uses an upload or upsert session selected by insert_strategy.

Key features

• exactly-once
• cdc
• support multiple table write
• timer flush

Options

name	type	required	default value	description
accessId	string	no	-	Aliyun AccessKey ID used to access MaxCompute.
accesskey	string	no	-	Aliyun AccessKey secret used to access MaxCompute.
sts_token	string	no	-	STS token used for temporary MaxCompute authentication. When sts_token is provided, accessId and accesskey are required.
endpoint	string	yes	-	MaxCompute endpoint, starting with http.
project	string	yes	-	MaxCompute project created in Alibaba Cloud.
table_name	string	yes	-	Target MaxCompute table name, for example fake.
schema_name	string	no	-	MaxCompute schema name (namespace between project and table). Required only when the table is in a non-default schema.
partition_spec	string	no	-	Partition spec for a MaxCompute partitioned table, for example ds='20220101'.
overwrite	boolean	no	false	Whether to overwrite the target table or partition.
schema_save_mode	enum	no	CREATE_SCHEMA_WHEN_NOT_EXIST	How to handle the target table before writing, such as RECREATE_SCHEMA or CREATE_SCHEMA_WHEN_NOT_EXIST.
data_save_mode	enum	no	APPEND_DATA	How to handle existing target data before writing, such as DROP_DATA, APPEND_DATA, or ERROR_WHEN_DATA_EXISTS.
custom_sql	string	no	-	Custom SQL to execute before writing when data_save_mode = CUSTOM_PROCESSING.
save_mode_create_template	string	no	see below	DDL template used when the sink creates the target table.
datetime_format	string	no	yyyy-MM-dd HH:mm:ss	Format string used to convert LocalDateTime fields to strings.
tunnel_endpoint	string	no	-	Custom endpoint URL for the MaxCompute Tunnel service. When not set, the endpoint is auto-inferred from the region.
tunnel_name	string	no	-	Tunnel Quota name used for exclusive resource groups. Requires both endpoint and tunnel_endpoint to be VPC endpoints.
insert_strategy	string	no	upload	Insert session strategy: upload uses an upload session, upsert uses an upsert session and requires a primary key.
multi_table_sink_replica	int	no	1	Number of sink writer replicas for each table in a multi-table job.
common-options		no	-	Sink plugin common parameters, such as plugin_input.

accessId [string]

accessId Your Maxcompute accessId that can access Alibaba Cloud.

accesskey [string]

accesskey Your Maxcompute accessKey that can access Alibaba Cloud.

sts_token [string]

sts_token Your MaxCompute STS Token for temporary authentication. Note: If sts_token is provided, accessId and accesskey are strictly required.

Passwordless Authentication (ECS RAM Role, Environment Variables, etc.) To use passwordless authentication seamlessly, simply leave accessId, accesskey, and sts_token all blank. The connector will automatically fall back to the Aliyun DefaultCredentialsProvider chain (Environment Variables, System Properties, CLI Profiles, OIDC, ECS RAM Roles).

endpoint [string]

endpoint Your Maxcompute endpoint start with http.

project [string]

project Your Maxcompute project which is created in Alibaba Cloud.

table_name [string]

table_name Target Maxcompute table name eg: fake.

partition_spec [string]

partition_spec This spec of Maxcompute partition table eg:ds='20220101'.

schema_name [string]

schema_name The MaxCompute Schema name (the namespace between Project and Table). Only required when the table resides in a non-default schema within your MaxCompute project. See Schema-related operations.

Default: not set (uses the project default schema).

overwrite [boolean]

overwrite Whether to overwrite the table or partition, default: false.

save_mode_create_template

We use templates to automatically create MaxCompute tables, which will create corresponding table creation statements based on the type of upstream data and schema type, and the default template can be modified according to the situation. Only work on multi-table mode at now.

Default template:

CREATE TABLE IF NOT EXISTS `${table}` (
${rowtype_fields}
) COMMENT '${comment}';

If a custom field is filled in the template, such as adding an id field

CREATE TABLE IF NOT EXISTS `${table}`
(   
    id,
    ${rowtype_fields}
) COMMENT '${comment}';

The connector will automatically obtain the corresponding type from the upstream to complete the filling, and remove the id field from rowtype_fields. This method can be used to customize the modification of field types and attributes.

You can use the following placeholders

• database: Used to get the database in the upstream schema
• table_name: Used to get the table name in the upstream schema
• rowtype_fields: Used to get all the fields in the upstream schema, we will automatically map to the field description of MaxCompute
• rowtype_primary_key: Used to get the primary key in the upstream schema (maybe a list)
• rowtype_unique_key: Used to get the unique key in the upstream schema (maybe a list)
• comment: Used to get the table comment in the upstream schema

schema_save_mode [Enum]

Before the synchronous task is turned on, different treatment schemes are selected for the existing surface structure of the target side.
Option introduction：
RECREATE_SCHEMA ：Will create when the table does not exist, delete and rebuild when the table is saved. If the partition_spec is set, the partition will be deleted and rebuilt.
CREATE_SCHEMA_WHEN_NOT_EXIST ：Will Created when the table does not exist, skipped when the table is saved. If the partition_spec is set, the partition will be created.
ERROR_WHEN_SCHEMA_NOT_EXIST ：Error will be reported when the table does not exist
IGNORE ：Ignore the treatment of the table

data_save_mode [Enum]

Before the synchronous task is turned on, different processing schemes are selected for data existing data on the target side.
Option introduction：
DROP_DATA： Preserve database structure and delete data
APPEND_DATA：Preserve database structure, preserve data
CUSTOM_PROCESSING：User defined processing
ERROR_WHEN_DATA_EXISTS：When there is data, an error is reported

custom_sql [String]

When data_save_mode selects CUSTOM_PROCESSING, you should fill in the CUSTOM_SQL parameter. This parameter usually fills in a SQL that can be executed. SQL will be executed before synchronization tasks.

datetime_format [String]

User-defined format string used to convert LocalDateTime fields to strings.

Use this option when you want to specify a custom datetime format that matches one of the predefined values in DateTimeUtils.Formatter (e.g. yyyy-MM-dd HH:mm:ss, yyyyMMddHHmmss, etc.).

Example values:

• yyyy-MM-dd HH:mm:ss
• yyyy-MM-dd HH:mm:ss.SSSSSS
• yyyy.MM.dd HH:mm:ss
• yyyy/MM/dd HH:mm:ss
• yyyy/M/d HH:mm
• yyyy-M-d HH:mm
• yyyy/M/d HH:mm:ss
• yyyy-M-d HH:mm:ss
• yyyyMMddHHmmss

Default: yyyy-MM-dd HH:mm:ss

tunnel_endpoint [String]

Specifies the custom endpoint URL for the MaxCompute Tunnel service.

By default, the endpoint is automatically inferred from the configured region.

This option allows you to override the default behavior and use a custom Tunnel endpoint. If not specified, the connector will use the region-based default Tunnel endpoint.

In general, you do not need to set tunnel_endpoint. It is only needed for custom networking, debugging, or local development.

Example values:

• https://dt.cn-hangzhou.maxcompute.aliyun.com
• https://dt.ap-southeast-1.maxcompute.aliyun.com
• http://maxcompute:8080

Default: Not set (auto-inferred from region)

tunnel_name [String]

tunnel_name Specifies the Tunnel Quota name for exclusive resource groups.

Tunnel Quota allows you to use dedicated computing resources for MaxCompute Tunnel data transfer, providing better performance and resource isolation.

Important: Tunnel Quota only works with VPC (Virtual Private Cloud) endpoints. It is not supported for public network access. You must configure both endpoint and tunnel_endpoint to use VPC endpoints when using tunnel_name.

If not specified, the default Tunnel quota will be used.

Example values:

• your_tunnel_quota_name

Default: Not set (use default quota)

insert_strategy [string]

If insert_strategy is set to upload, insert operations use an upload session. If set to upsert, insert operations use an upsert session. Upsert sessions require a primary key.

Note: Using upload sessions for insert operations alongside update or delete operations may cause insert records to appear in the table later than expected. When a primary key is present, it is recommended to set insert_strategy to upsert to ensure consistent upsert behavior.

UPDATE_AFTER and DELETE rows are always written through a MaxCompute upsert session, so the target table must have a primary key when the job contains update or delete rows. UPDATE_BEFORE rows are not supported by this sink.

multi_table_sink_replica [int]

The number of writer replicas in multi-table sink mode. The default value is 1.

Use this option when upstream data contains multiple table identifiers and table_name uses placeholders such as ${table_name}. For example, table_name = "${table_name}_sink" writes upstream table test_table to target table test_table_sink.

common options

Sink plugin common parameters, please refer to Sink Common Options for details.

Examples

Append Data

sink {
  Maxcompute {
    accessId="<your access id>"
    accesskey="<your access Key>"
    endpoint="<http://service.odps.aliyun.com/api>"
    project="<your project>"
    table_name="<your table name>"
    #partition_spec="<your partition spec>"
    #overwrite = false
  }
}

Multiple Tables

source {
  FakeSource {
    tables_configs = [
      {
        schema = {
          table = "test_table"
          fields {
            ID = int
            NAME = string
            AGE = int
          }
          primaryKey {
            name = "ID"
            columnNames = [ID]
          }
        }
        rows = [
          { kind = INSERT, fields = [1, "INSERT_TEST1", 20] }
          { kind = INSERT, fields = [2, "INSERT_TEST2", 30] }
        ]
      },
      {
        schema = {
          table = "test_table_2"
          fields {
            ID = int
            NAME = string
            AGE = int
          }
          primaryKey {
            name = "ID"
            columnNames = [ID]
          }
        }
        rows = [
          { kind = INSERT, fields = [1, "INSERT_TEST1", 20] }
        ]
      }
    ]
  }
}

sink {
  Maxcompute {
    accessId = "ak"
    accesskey = "sk"
    endpoint = "http://maxcompute:8080"
    tunnel_endpoint = "http://maxcompute:8080"
    project = "mocked_mc"
    table_name = "${table_name}_sink"
    insert_strategy = "upsert"
    multi_table_sink_replica = 1
  }
}

Upsert or Delete Rows

Use insert_strategy = "upsert" when the upstream schema has a primary key and the job contains update or delete rows. The example below uses an update row; delete rows use the same sink settings.

source {
  FakeSource {
    tables_configs = [
      {
        schema = {
          table = "test_table_sink"
          fields {
            ID = int
            NAME = string
            AGE = int
          }
          primaryKey {
            name = "ID"
            columnNames = [ID]
          }
        }
        rows = [
          {
            kind = UPDATE_AFTER
            fields = [1, "UPSERT_TEST", 100]
          }
        ]
      }
    ]
  }
}

sink {
  Maxcompute {
    accessId = "ak"
    accesskey = "sk"
    endpoint = "http://maxcompute:8080"
    tunnel_endpoint = "http://maxcompute:8080"
    project = "mocked_mc"
    table_name = "test_table_sink"
    insert_strategy = "upsert"
  }
}

Changelog

Change Log

Change	Commit	Version
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)	https://github.com/apache/seatunnel/commit/9212a77140	2.3.12
[Bug][Connector-V2] NoSuchMethodError caused by Netty version conflict on Spark 3.3.0 (#9632)	https://github.com/apache/seatunnel/commit/4d2b55ce3c	2.3.12
[Improve][Connector-V2] Replace deprecated createDownloadSession by buildDownloadSession (#9555)	https://github.com/apache/seatunnel/commit/6862945eef	2.3.12
[Improve][Connector-V2] Add tunnel_endpoint option to MaxCompute source for emulator test (#9548)	https://github.com/apache/seatunnel/commit/b3f3c527ca	2.3.12
[Improve][Connector-V2] Support maxcompute sink writer upsert/delete action with upsert session mode (#9462)	https://github.com/apache/seatunnel/commit/eb9c8704b9	2.3.12
[Improve][Connector-V2] Support maxcompute sink writer with timestamp field type (#9234)	https://github.com/apache/seatunnel/commit/a513c495e3	2.3.12
[Feature][Transform] Support define sink column type (#9114)	https://github.com/apache/seatunnel/commit/ab7119e507	2.3.11
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)	https://github.com/apache/seatunnel/commit/4f5adeb1c7	2.3.11
[Improve] maxcompute options (#9163)	https://github.com/apache/seatunnel/commit/fdacbae1af	2.3.11
[Fix][Connector-V2] Fix maxcompute write with multi parallelism (#9089)	https://github.com/apache/seatunnel/commit/9426b7ba2c	2.3.11
[Fix][Connector-V2] Fix maxcompute sink write date less than actual date (#8999)	https://github.com/apache/seatunnel/commit/fc942a599b	2.3.11
[Fix][Connector-V2] Fix maxcompute read with partition spec (#8896)	https://github.com/apache/seatunnel/commit/e62bf6c65c	2.3.10
[Fix][Connector-V2] Fix MaxCompute cannot get project and tableName when use schema (#8865)	https://github.com/apache/seatunnel/commit/a24fa8fef6	2.3.10
[Improve] restruct connector common options (#8634)	https://github.com/apache/seatunnel/commit/f3499a6eeb	2.3.10
[Feature][Connector-V2] Support maxcompute source with multi-table (#8582)	https://github.com/apache/seatunnel/commit/0f78242923	2.3.10
[Fix][Connector-V2] Fixed adding table comments (#8514)	https://github.com/apache/seatunnel/commit/edca75b0d6	2.3.10
[Improve][Connector-V2] MaxComputeSink support create partition in savemode (#8474)	https://github.com/apache/seatunnel/commit/0b8f9de465	2.3.10
[Improve][Transform] Rename sql transform table name from 'fake' to 'dual' (#8298)	https://github.com/apache/seatunnel/commit/e6169684fb	2.3.9
[Feature][Connector-V2] Support MaxCompute save mode (#8277)	https://github.com/apache/seatunnel/commit/44ea675f1e	2.3.9
[Improve][dist]add shade check rule (#8136)	https://github.com/apache/seatunnel/commit/51ef800016	2.3.9
[Feature][Core] Rename result_table_name/source_table_name to plugin_input/plugin_output (#8072)	https://github.com/apache/seatunnel/commit/c7bbd322db	2.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
[Fix] Fix dead link on seatunnel connectors list url (#7453)	https://github.com/apache/seatunnel/commit/62b4f16f4e	2.3.8
[BugFix][Connector-V2][Maxcompute]fix:Maxcompute sink can't map field(#7164) (#7168)	https://github.com/apache/seatunnel/commit/d5abf8f506	2.3.6
[Feature] Add unsupported datatype check for all catalog (#5890)	https://github.com/apache/seatunnel/commit/b9791285a0	2.3.4
FakeSource support generate different CatalogTable for MultipleTable (#5766)	https://github.com/apache/seatunnel/commit/a8b93805ea	2.3.4
[Improve][Common] Introduce new error define rule (#5793)	https://github.com/apache/seatunnel/commit/9d1b2582b2	2.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)	https://github.com/apache/seatunnel/commit/8de7408100	2.3.4
[Improve][Connector] Add field name to DataTypeConvertor to improve error message (#5782)	https://github.com/apache/seatunnel/commit/ab60790f0d	2.3.4
[Improve][Test] Move MaxCompute test case file (#5786)	https://github.com/apache/seatunnel/commit/38132f5158	2.3.4
[Fix] Fix MaxCompute use not exist SCHEMA option (#5708)	https://github.com/apache/seatunnel/commit/ba4782a67d	2.3.4
[Feature] Support catalog in MaxCompute Source (#5283)	https://github.com/apache/seatunnel/commit/946d89cb95	2.3.4
[Bugfix][Connector-V2][maxcompute] sink commit with Block not exsits on server (#4725)	https://github.com/apache/seatunnel/commit/2760cae73c	2.3.2
[Bug][Maxcompute] Fix failed to parse some maxcompute type (#3894)	https://github.com/apache/seatunnel/commit/642901f0a2	2.3.1
[Improve][build] Give the maven module a human readable name (#4114)	https://github.com/apache/seatunnel/commit/d7cd601051	2.3.1
[Improve][Project] Code format with spotless plugin. (#4101)	https://github.com/apache/seatunnel/commit/a2ab166561	2.3.1
[Feature][Connector] add get source method to all source connector (#3846)	https://github.com/apache/seatunnel/commit/417178fb84	2.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)	https://github.com/apache/seatunnel/commit/b9164b8ba1	2.3.1
[Feature][Connector-V2][Maxcompute] Add Maxcompute source & sink connector (#3640)	https://github.com/apache/seatunnel/commit/80cf8f4e42	2.3.0