MaxCompute:Use Flink to write data to MaxCompute

Prerequisites

Before you begin, ensure that you have:

• A running open source Flink cluster (version 1.13, 1.15, 1.16, or 1.17)

• A MaxCompute project with permission to create tables and write data

• Your AccessKey ID and AccessKey secret (available on the AccessKey Pair page)

Step 1: Create a MaxCompute table

Create the target Delta table in MaxCompute. The following example creates a non-partitioned table and a partitioned table. For Delta table property details, see Delta table parameters.

-- Non-partitioned Delta table
CREATE TABLE mf_flink_tt (
  id BIGINT NOT NULL,
  name STRING,
  age INT,
  status BOOLEAN,
  PRIMARY KEY (id)
)
TBLPROPERTIES (
  "transactional" = "true",
  "write.bucket.num" = "64",
  "acid.data.retain.hours" = "12"
);

-- Partitioned Delta table
CREATE TABLE mf_flink_tt_part (
  id BIGINT NOT NULL,
  name STRING,
  age INT,
  status BOOLEAN,
  PRIMARY KEY (id)
)
PARTITIONED BY (dd STRING, hh STRING)
TBLPROPERTIES (
  "transactional" = "true",
  "write.bucket.num" = "64",
  "acid.data.retain.hours" = "12"
);

Step 2: Install the Flink connector

Download the JAR for your Flink version and add it to the cluster.

The 1.16 connector is compatible with both Flink 1.16 and 1.17.

Move the downloaded JAR to the lib directory of your Flink installation:

mv flink-connector-odps-1.13-shaded.jar $FLINK_HOME/lib/

Step 3: Start Flink

cd $FLINK_HOME/bin
./start-cluster.sh
./sql-client.sh

Step 4: Create a Flink table and write data

Choose Flink SQL or the DataStream API based on your job type.

-- Register the non-partitioned table
CREATE TABLE mf_flink (
  id BIGINT,
  name STRING,
  age INT,
  status BOOLEAN,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'          = 'maxcompute',
  'table.name'         = 'mf_flink_tt',
  'sink.operation'     = 'upsert',
  'odps.access.id'     = 'LTAI****************',
  'odps.access.key'    = '********************',
  'odps.end.point'     = 'http://service.cn-beijing.maxcompute.aliyun.com/api',
  'odps.project.name'  = 'mf_mc_bj'
);

-- Register the partitioned table
CREATE TABLE mf_flink_part (
  id BIGINT,
  name STRING,
  age INT,
  status BOOLEAN,
  dd STRING,
  hh STRING,
  PRIMARY KEY (id) NOT ENFORCED
) PARTITIONED BY (`dd`, `hh`)
WITH (
  'connector'          = 'maxcompute',
  'table.name'         = 'mf_flink_tt_part',
  'sink.operation'     = 'upsert',
  'odps.access.id'     = 'LTAI****************',
  'odps.access.key'    = '********************',
  'odps.end.point'     = 'http://service.cn-beijing.maxcompute.aliyun.com/api',
  'odps.project.name'  = 'mf_mc_bj'
);

-- Insert into the non-partitioned table
INSERT INTO mf_flink VALUES (1, 'Danny', 27, false);

-- Query MaxCompute to verify
SELECT * FROM mf_flink_tt;
-- Result:
-- +----+-------+-----+--------+
-- | id | name  | age | status |
-- +----+-------+-----+--------+
-- | 1  | Danny |  27 | false  |
-- +----+-------+-----+--------+

-- Update the record via upsert (same primary key, new age value)
INSERT INTO mf_flink VALUES (1, 'Danny', 28, false);

SELECT * FROM mf_flink_tt;
-- Result:
-- +----+-------+-----+--------+
-- | id | name  | age | status |
-- +----+-------+-----+--------+
-- | 1  | Danny |  28 | false  |
-- +----+-------+-----+--------+

-- Insert into the partitioned table
INSERT INTO mf_flink_part VALUES (1, 'Danny', 27, false, '01', '01');

SELECT * FROM mf_flink_tt_part WHERE dd = '01' AND hh = '01';
-- Result:
-- +----+-------+-----+--------+----+----+
-- | id | name  | age | status | dd | hh |
-- +----+-------+-----+--------+----+----+
-- | 1  | Danny |  27 | false  | 01 | 01 |
-- +----+-------+-----+--------+----+----+

<dependency>
  <groupId>com.aliyun.odps</groupId>
  <artifactId>flink-connector-maxcompute</artifactId>
  <version>xxx</version>
  <scope>system</scope>
  <systemPath>${mvn_project.basedir}/lib/flink-connector-maxcompute-xxx-shaded.jar</systemPath>
</dependency>

package com.aliyun.odps.flink.examples;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.odps.table.OdpsOptions;
import org.apache.flink.odps.util.OdpsConf;
import org.apache.flink.odps.util.OdpsPipeline;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.data.RowData;

public class Examples {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(120 * 1000);

        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env);

        Table source = streamTableEnvironment.sqlQuery("SELECT * FROM source_table");
        DataStream<RowData> input = streamTableEnvironment.toAppendStream(source, RowData.class);

        // Configure sink behavior
        Configuration config = new Configuration();
        config.set(OdpsOptions.SINK_OPERATION, "upsert");
        config.set(OdpsOptions.UPSERT_COMMIT_THREAD_NUM, 8);
        config.set(OdpsOptions.UPSERT_MAJOR_COMPACT_MIN_COMMITS, 100);

        // Configure MaxCompute credentials and endpoint
        OdpsConf odpsConfig = new OdpsConf(
            "accessid",
            "accesskey",
            "endpoint",
            "project",
            "tunnel endpoint"
        );

        // Build and run the sink pipeline
        OdpsPipeline.Builder builder = OdpsPipeline.builder();
        builder.projectName("sql2_isolation_2a")
               .tableName("user_ledger_portfolio")
               .partition("")
               .configuration(config)
               .odpsConf(odpsConfig)
               .sink(input, false);

        env.execute();
    }
}

Prerequisites

Before you begin, ensure that you have:

• Access to the Realtime Compute for Apache Flink console

• A MaxCompute project with permission to create tables and write data

• Your AccessKey ID and AccessKey secret (available on the AccessKey Pair page)

Step 1: Create a MaxCompute table

SET odps.sql.type.system.odps2 = true;

DROP TABLE IF EXISTS mf_flink_upsert;

CREATE TABLE mf_flink_upsert (
  c1 INT NOT NULL,
  c2 STRING,
  gt TIMESTAMP,
  PRIMARY KEY (c1)
)
PARTITIONED BY (ds STRING)
TBLPROPERTIES (
  "transactional" = "true",
  "write.bucket.num" = "64",
  "acid.data.retain.hours" = "12"
);

Step 2: Create and deploy a Flink SQL job

On the Flink job development page, create a Flink SQL job. The following example defines a source table that auto-generates random data and a sink table that writes to MaxCompute in upsert mode. For job development details, see Job development map.

-- Source table with auto-generated data
CREATE TEMPORARY TABLE fake_src_table (
  c1 INT,
  c2 VARCHAR,
  gt AS CURRENT_TIMESTAMP
) WITH (
  'connector'               = 'faker',
  'fields.c2.expression'    = '#{superhero.name}',
  'rows-per-second'         = '100',
  'fields.c1.expression'    = '#{number.numberBetween ''0'',''1000''}'
);

-- Sink table pointing to the MaxCompute Delta table
CREATE TEMPORARY TABLE test_c_d_g (
  c1 INT,
  c2 VARCHAR,
  gt TIMESTAMP,
  ds VARCHAR,
  PRIMARY KEY (c1) NOT ENFORCED
) PARTITIONED BY (ds)
WITH (
  'connector'               = 'maxcompute',
  'table.name'              = 'mf_flink_upsert',
  'sink.operation'          = 'upsert',
  'odps.access.id'          = 'LTAI****************',
  'odps.access.key'         = '********************',
  'odps.end.point'          = 'http://service.cn-beijing.maxcompute.aliyun.com/api',
  'odps.project.name'       = 'mf_mc_bj',
  'upsert.write.bucket.num' = '64'
);

-- Write logic
INSERT INTO test_c_d_g
SELECT
  c1,
  c2,
  gt,
  date_format(gt, 'yyyyMMddHH') AS ds
FROM fake_src_table;

For odps.end.point, use the internal network endpoint for your region. The upsert.write.bucket.num value must match the write.bucket.num property of the MaxCompute table.

After you develop the job, deploy it from the Flink job development page.

Step 3: Verify data in MaxCompute

SELECT * FROM mf_flink_upsert WHERE ds = '2023061517';

-- The Flink source generates random data, so your results will differ.
-- Example output:
-- +-----+--------------------+-------------------------+------------+
-- | c1  | c2                 | gt                      | ds         |
-- +-----+--------------------+-------------------------+------------+
-- | 0   | Skaar              | 2023-06-16 01:59:41.116 | 2023061517 |
-- | 21  | Supah Century      | 2023-06-16 01:59:59.117 | 2023061517 |
-- | 104 | Dark Gorilla Grodd | 2023-06-16 01:59:57.117 | 2023061517 |
-- | 126 | Leader             | 2023-06-16 01:59:39.116 | 2023061517 |
-- +-----+--------------------+-------------------------+------------+

Basic parameters

Partition parameters

FileCached mode parameters

Use FileCached mode when you have a large number of dynamic partitions. In this mode, data is written to local cache files before being uploaded to MaxCompute.