Realtime Compute for Apache Flink:Big Data Computing Service MaxCompute

Parameter

Description

Data type

Required

Default value

Remarks

connector

The table type.

String

Yes

None

The value must be odps.

endpoint

The endpoint of the MaxCompute service.

String

Yes

None

For more information, see Endpoints.

tunnelEndpoint

The endpoint of the MaxCompute Tunnel service.

String

No

None

For more information, see Endpoints.

project

The name of the MaxCompute project.

String

Yes

None

None.

schemaName

The name of the MaxCompute schema.

String

No

None

This parameter is required only if the schema feature is enabled for the MaxCompute project. Set this parameter to the name of the schema to which the MaxCompute table belongs. For more information, see Schema operations.

tableName

The name of the MaxCompute table.

String

Yes

None

None.

accessId

The AccessKey ID for MaxCompute.

String

Yes

None

For more information, see How do I view the AccessKey ID and AccessKey secret?

accessKey

The AccessKey secret for MaxCompute.

String

Yes

None

partition

The name of the MaxCompute partition.

String

No

None

This parameter is not required for non-partitioned tables and incremental source tables.

compressAlgorithm

The compression algorithm used by MaxCompute Tunnel.

String

No

SNAPPY

Valid values:

• RAW (no compression)

• ZLIB

• SNAPPY

  SNAPPY provides significantly higher throughput than ZLIB. In test scenarios, throughput increases by about 50%.

quotaName

The quota name of the dedicated resource for MaxCompute Data Transmission Service.

String

No

None

Set this parameter to use the dedicated MaxCompute Data Transmission Service.

Parameter

Description

Data type

Required

Default value

Remarks

maxPartitionCount

The maximum number of partitions that can be read.

Integer

No

100

If the number of partitions to read exceeds this value, the error The number of matched partitions exceeds the default limit occurs.

useArrow

Specifies whether to use the Arrow format to read data.

Boolean

No

false

Using the Arrow format lets you call the MaxCompute Storage API.

splitSize

The size of data to pull at a time when reading data in the Arrow format.

MemorySize

No

256 MB

This parameter is supported only in VVR 8.0.8 and later.

compressCodec

The compression algorithm to use when reading data in the Arrow format.

String

No

""

Valid values:

• "" (no compression)

• ZSTD

• LZ4_FRAME

Specifying a compression algorithm can improve throughput compared to no compression.

dynamicLoadBalance

Specifies whether to allow dynamic sharding.

Boolean

No

false

Valid values:

• true: Allowed

• false: The operation is not allowed.

Allowing dynamic sharding lets you leverage the processing power of different Flink nodes and reduces the overall read time for the source table. However, it can lead to inconsistent data volumes read by different nodes, causing data skew.

Parameter

Description

Data type

Required

Default value

Remarks

startPartition

The starting MaxCompute partition (inclusive) for incremental reading.

String

Yes

None

• Using this parameter enables the incremental source table mode and ignores the `partition` parameter.

• For multi-level partitions, you must declare the value of each partition key column from the highest level to the lowest.

subscribeIntervalInSec

The interval at which to poll MaxCompute for the partition list.

Integer

No

30

The unit is seconds.

modifiedTableOperation

The action to take when partition data is modified during reading.

Enum (NONE, SKIP)

No

NONE

The download session is saved in a checkpoint. Each time the job recovers from a checkpoint, it tries to resume reading from that session. If the session is unavailable because the partition data was modified, the Flink job enters a restart loop. You can set this parameter to handle this situation:

• NONE: You must modify the `startPartition` parameter to be greater than the unavailable partition and start the job without a state.

• SKIP: If you do not want to start the job without a state, set the mode to SKIP. When Flink tries to recover the session from a checkpoint, it skips the unavailable partition.

Parameter

Description

Data type

Required

Default value

Remarks

useStreamTunnel

Specifies whether to use MaxCompute Stream Tunnel to upload data.

Boolean

No

false

flushIntervalMs

The flush interval for the MaxCompute Tunnel Writer buffer.

Long

No

30000 (30 seconds)

Data is first written to a buffer. When the buffer is full or the `flushIntervalMs` interval is reached, the data is written to the destination table in batches.

• Stream Tunnel: Data is visible immediately after being flushed.

• Batch Tunnel: Data is visible only after a checkpoint is completed. Set `flushIntervalMs` to 0 to disable timed flushes and avoid delays.

The unit is milliseconds.

batchSize

The flush size of the MaxCompute Tunnel Writer buffer.

Long

No

67108864 (64 MB)

The unit is bytes.

When a record is written, the data is first stored in the MaxCompute buffer. When the buffer reaches a certain size (`batchSize`), the data in the buffer is written to the destination MaxCompute table.

numFlushThreads

The number of threads for flushing the MaxCompute Tunnel Writer buffer.

Integer

No

1

slotNum

The number of slots used by the MaxCompute Tunnel Writer.

Integer

No

0

For more information about the limits on the number of slots, see Data Transmission Service overview.

dynamicPartitionLimit

The maximum number of dynamic partitions to write to.

Integer

No

100

If the number of dynamic partitions written to the sink table between two checkpoints exceeds `dynamicPartitionLimit`, the error Too many dynamic partitions occurs.

retryTimes

The maximum number of retries for requests to the MaxCompute server.

Integer

No

3

When creating a session, committing a session, or flushing data, the MaxCompute service may be temporarily unavailable. The system retries based on this configuration.

sleepMillis

The retry interval.

Integer

No

1000

The unit is milliseconds.

enableUpsert

Specifies whether to use MaxCompute Upsert Tunnel to upload data.

Boolean

No

false

Valid values:

• true: Uses Upsert Tunnel to process INSERT, UPDATE_AFTER, and DELETE data from Flink.

• false: Uses Batch Tunnel or Stream Tunnel based on the useStreamTunnel parameter to process INSERT and UPDATE_AFTER data from Flink.

upsertAsyncCommit

Specifies whether to use asynchronous mode when committing a session in Upsert mode.

Boolean

No

false

Valid values:

• true: Uses asynchronous mode. The commit takes less time, but the written data is not immediately readable upon completion.

• false: Uses synchronous mode by default. The system waits for the server-side to process the session before committing.

upsertCommitTimeoutMs

The timeout period for committing a session in Upsert mode.

Integer

No

120000

(120 seconds)

The unit is milliseconds.

sink.operation

The write mode when writing to a Delta Lake table.

String

No

insert

Valid values:

• insert: Appends data.

• upsert: Updates data.

sink.parallelism

The degree of parallelism for writing to a Delta Lake table.

Integer

No

None

• The degree of parallelism for writing. If not set, the upstream data parallelism is used by default.

• This parameter is supported only in VVR 8.0.10 and later.

sink.file-cached.enable

Specifies whether to use file cache mode when writing to dynamic partitions of a Delta Lake table.

Boolean

No

false

Valid values:

• true: Uses file cache mode.

• false: Does not use file cache mode.

Using file cache mode reduces the number of small files written to the server-side, but increases the latency of data output. When the degree of parallelism for the sink table is high, use file cache mode.

sink.file-cached.writer.num

The number of concurrent data uploads for a single task in file cache mode.

Integer

No

16

• This parameter takes effect only if sink.file-cached.enable is set to true.

• Do not significantly increase this value. Writing to too many partitions at the same time can easily cause an out-of-memory (OOM) error.

sink.bucket.check-interval

The interval for checking the file size in file cache mode. The unit is milliseconds (ms).

Integer

No

60000

This parameter takes effect only if sink.file-cached.enable is set to true.

sink.file-cached.rolling.max-size

The maximum size of a single cache file in file cache mode.

MemorySize

No

16 M

• This parameter takes effect only if sink.file-cached.enable is set to true.

• If the file size exceeds this value, the file data is uploaded to the server-side.

sink.file-cached.memory

The maximum off-heap memory size used for writing files in file cache mode.

MemorySize

No

64 MB

This parameter takes effect only if sink.file-cached.enable is set to true.

sink.file-cached.memory.segment-size

The buffer size used for writing files in file cache mode.

MemorySize

No

128 KB

This parameter takes effect only if sink.file-cached.enable is set to true.

sink.file-cached.flush.always

Specifies whether to use a cache when writing files in file cache mode.

Boolean

No

true

This parameter takes effect only if sink.file-cached.enable is set to true.

sink.file-cached.write.max-retries

The number of retries for uploading data in file cache mode.

Integer

No

3

This parameter takes effect only if sink.file-cached.enable is set to true.

upsert.writer.max-retries

The number of retries after an Upsert Writer fails to write to a bucket.

Integer

No

3

upsert.writer.buffer-size

The cache size for a single Upsert Writer in Flink.

MemorySize

No

64 m

• When the total buffer size of all buckets reaches the preset threshold, the system automatically triggers a flush operation to update the data to the server-side.

• A single upsert writer writes to multiple buckets simultaneously. Increase this value to improve write efficiency.

• If many partitions are written, there is a risk of an OOM error. Consider reducing this parameter value.

upsert.writer.bucket.buffer-size

The cache size for a single bucket in Flink.

MemorySize

No

1 m

When cluster memory resources are tight, you can reduce this parameter value.

upsert.write.bucket.num

The number of buckets in the table to be written to.

Integer

Yes

None

This value must be the same as the value of the write.bucket.num property of the Delta Lake table to be written to.

upsert.write.slot-num

The number of Tunnel slots used by a single session.

Integer

No

1

upsert.commit.max-retries

The number of retries for an Upsert Session Commit.

Integer

No

3

upsert.commit.thread-num

The degree of parallelism for an Upsert Session Commit.

Integer

No

16

Do not set this parameter to a very large value. A higher number of concurrent commits consumes more resources, which may lead to performance issues or excessive resource consumption.

upsert.commit.timeout

The timeout period for an Upsert Session Commit. The unit is seconds (s).

Integer

No

600

upsert.flush.concurrent

The maximum number of buckets that can be written to simultaneously for a single partition.

Integer

No

2

Each time data in a bucket is flushed, it consumes a Tunnel Slot resource.

insert.commit.thread-num

The degree of parallelism for a Commit Session.

Integer

No

16

insert.arrow-writer.enable

Specifies whether to use the Arrow format.

Boolean

No

false

Valid values:

• true: Uses Arrow mode.

• false: Does not use Arrow mode.

insert.arrow-writer.batch-size

The maximum number of rows in an Arrow Batch.

Integer

No

512

insert.arrow-writer.flush-interval

The Writer flush interval. The unit is milliseconds (ms).

Integer

No

100000

insert.writer.buffer-size

The cache size of the Buffered Writer.

MemorySize

No

64 MB

upsert.partial-column.enable

Specifies whether to update only some columns.

Boolean

No

false

This parameter takes effect only when the sink table is a Delta Lake table. For more information, see Partial column update.

Valid values:

• true: Updates only some columns.

• false: Updates all columns.

Data is written in different ways depending on whether a primary key for the update data exists in the sink table:

• If data with the same primary key exists in the sink table, the record is updated by its primary key. The update uses the values from the specified columns that are not null.

• If no data with the same primary key exists in the sink table, a new record is added with the primary key. The values of the specified columns are inserted, and null is inserted for all other columns.

Parameter

Description

Data type

Required

Default value

Remarks

cache

The cache policy.

String

Yes

None

Currently, MaxCompute dimension tables support only the ALL policy, which must be explicitly declared. This policy is suitable for scenarios where the remote table is small and has many MISS KEYs (where the ON condition fails to associate data during a join between the source table and the dimension table).

ALL: Caches all data in the dimension table. Before the job runs, the system loads all data from the dimension table into the cache. All subsequent dimension table queries are performed through the cache. If data is not found in the cache, the key does not exist. The full cache is reloaded after it expires.

cacheSize

The maximum number of data records to cache.

Long

No

100000

If the dimension table data volume exceeds `cacheSize`, the error Row count of table <table-name> partition <partition-name> exceeds maxRowCount limit occurs.

cacheTTLMs

The cache timeout period, which is the cache update interval.

Long

No

Long.MAX_VALUE (never updates)

The unit is milliseconds.

cacheReloadTimeBlackList

The disallowed cache refresh time. The cache is not updated during the time period specified by this parameter.

String

No

None

Prevents the cache from being updated during critical periods, such as traffic peaks during a promotion, which could cause job instability. For more information about how to configure this parameter, see How do I configure the cacheReloadTimeBlackList parameter?.

maxLoadRetries

The maximum number of attempts to update the cache, including the initial data pull at job startup. If this number is exceeded, the job fails.

Integer

No

10

None.