MaxCompute:Menulis data ke MaxCompute dengan Flink

Latar Belakang

• Mode tulis yang didukung

  Konektor Flink mendukung dua mode tulis: upsert dan insert. Dalam mode upsert, Anda dapat mengelompokkan aliran data dengan salah satu cara berikut:

  • Kelompokkan berdasarkan primary key

  • Kelompokkan berdasarkan bidang partisi

    Meskipun cocok untuk jumlah partisi yang besar, pengelompokan berdasarkan bidang partisi dapat menyebabkan kesenjangan data.

• Untuk prosedur penulisan upsert dan parameter yang direkomendasikan, lihat Ingest data real-time ke gudang data.

• Tentukan mode tulis menggunakan parameter konektor Flink. Untuk daftar lengkap parameter konektor, lihat Lampiran: Parameter konektor Flink.

• Atur interval checkpoint untuk pekerjaan tulis upsert Flink minimal 3 menit. Interval yang lebih pendek dapat mengurangi efisiensi penulisan dan menghasilkan banyak file kecil.

• Tabel berikut memetakan tipe data antara Realtime Compute for Apache Flink dan MaxCompute.

  Tipe data Flink	Tipe data MaxCompute
  CHAR(p)	CHAR(p)
  VARCHAR(p)	VARCHAR(p)
  STRING	STRING
  BOOLEAN	BOOLEAN
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  INT	INT
  BIGINT	BIGINT
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DECIMAL(p, s)	DECIMAL(p, s)
  DATE	DATE
  TIMESTAMP(9) WITHOUT TIME ZONE, TIMESTAMP_LTZ(9)	TIMESTAMP
  TIMESTAMP(3) WITHOUT TIME ZONE, TIMESTAMP_LTZ(3)	DATETIME
  BYTES	BINARY
  ARRAY<T>	ARRAY<T>
  MAP<K, V>	MAP<K, V>
  ROW	STRUCT

  Catatan

  Tipe data TIMESTAMP Flink tidak mencakup informasi zona waktu, sedangkan tipe data TIMESTAMP MaxCompute mencakupnya. Perbedaan ini dapat menyebabkan selisih waktu 8 jam. Untuk menyelaraskan timestamp, gunakan TIMESTAMP_LTZ(9).

  -- Flink SQL
  CREATE TEMPORARY TABLE odps_source(
    id BIGINT NOT NULL COMMENT 'ID',
    created_time TIMESTAMP NOT NULL COMMENT 'Waktu pembuatan',
    updated_time TIMESTAMP_LTZ(9) NOT NULL COMMENT 'Waktu pembaruan',
  PRIMARY KEY (id) NOT ENFORCED
  ) WITH (
  'connector' = 'maxcompute',
  ...
  );

Menulis data dari kluster Flink yang dikelola sendiri

1. Prasyarat: Buat tabel MaxCompute.

   Anda harus terlebih dahulu membuat tabel MaxCompute sebagai tujuan penulisan data Flink. Contoh berikut menunjukkan proses ini dengan membuat dua tabel (tabel delta non-partisi dan tabel partisi). Untuk informasi tentang pengaturan properti tabel, lihat parameter tabel delta.

   -- Buat tabel delta non-partisi.
   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") ;
   
   --Buat tabel delta partisi.
   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") ;
   

2. Siapkan kluster Flink open source. Konektor ini mendukung Flink 1.13, 1.15, 1.16, dan 1.17. Pilih konektor Flink yang sesuai dengan versi Flink Anda:

   • Konektor Flink untuk Flink 1.13

   • Konektor Flink untuk Flink 1.15

   • Konektor Flink untuk Flink 1.16 dan 1.17

   Catatan

   • Konektor Flink untuk Flink 1.16 kompatibel dengan Flink 1.17.

   • Topik ini menggunakan konektor Flink untuk Flink 1.13 sebagai contoh. Unduh dan ekstrak paket tersebut.

3. Unduh konektor Flink dan tambahkan ke paket kluster Flink Anda.

   1. Unduh paket JAR konektor Flink ke lingkungan lokal Anda.

   2. Tambahkan paket JAR konektor Flink ke direktori lib dari paket instalasi Flink yang telah diekstrak.

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

4. Jalankan layanan instans Flink.

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

5. Jalankan klien Flink.

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

6. Buat tabel Flink dan konfigurasikan parameter konektor Flink.

   Anda dapat membuat tabel Flink dan mengonfigurasi parameternya menggunakan Flink SQL atau DataStream API. Bagian berikut menyediakan contoh inti untuk kedua pendekatan tersebut.

   Flink SQL

   1. Di editor Flink SQL, jalankan perintah berikut untuk membuat tabel dan mengonfigurasi parameter.

      -- Daftarkan tabel non-partisi di Flink SQL.
      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'
      );
      
      -- Daftarkan tabel partisi di Flink SQL.
      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'
      );

   2. Tulis data ke tabel Flink dan kueri tabel MaxCompute untuk memverifikasi hasilnya.

      -- Masukkan data ke tabel non-partisi di klien Flink SQL.
      INSERT INTO mf_flink VALUES (1,'Danny',27, false);
      
      -- Hasil kueri di MaxCompute.
      SELECT * FROM mf_flink_tt;
      +------------+------+------+--------+
      | id         | name | age  | status |
      +------------+------+------+--------+
      | 1          | Danny | 27   | false  |
      +------------+------+------+--------+
      
      -- Masukkan data ke tabel non-partisi di klien Flink SQL untuk memperbarui catatan.
      INSERT INTO mf_flink VALUES (1,'Danny',28, false);
      -- Hasil kueri di MaxCompute.
      SELECT * FROM mf_flink_tt;
      +------------+------+------+--------+
      | id         | name | age  | status |
      +------------+------+------+--------+
      | 1          | Danny | 28   | false  |
      +------------+------+------+--------+
      
      -- Masukkan data ke tabel partisi di klien Flink SQL.
      INSERT INTO mf_flink_part VALUES (1,'Danny',27, false, '01','01');
      -- Hasil kueri di MaxCompute.
      SELECT * FROM mf_flink_tt_part WHERE dd=01 AND hh=01;
      +------------+------+------+--------+----+----+
      | id         | name | age  | status | dd | hh |
      +------------+------+------+--------+----+----+
      | 1          | Danny | 27   | false  | 01 | 01 |
      +------------+------+------+--------+----+----+
      
      -- Masukkan data ke tabel partisi di klien Flink SQL untuk memperbarui catatan.
      INSERT INTO mf_flink_part VALUES (1,'Danny',30, false, '01','01');
      -- Hasil kueri di MaxCompute.
      SELECT * FROM mf_flink_tt_part WHERE dd=01 AND hh=01;
      +------------+------+------+--------+----+----+
      | id         | name | age  | status | dd | hh |
      +------------+------+------+--------+----+----+
      | 1          | Danny | 30   | false  | 01 | 01 |
      +------------+------+------+--------+----+----+

   DataStream API

   1. Untuk menggunakan DataStream API, tambahkan dependensi berikut.

      <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>

      Catatan

      Ganti "xxx" dengan nomor versi aktual.

   2. Kode contoh berikut menunjukkan cara membuat tabel dan mengonfigurasi parameter.

      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);
      
              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);
      
              OdpsConf odpsConfig = new OdpsConf("accessid",
                      "accesskey",
                      "endpoint",
                      "project",
                      "tunnel endpoint");
      
              OdpsPipeline.Builder builder = OdpsPipeline.builder();
              builder.projectName("sql2_isolation_2a")
                      .tableName("user_ledger_portfolio")
                      .partition("")
                      .configuration(config)
                      .odpsConf(odpsConfig)
                      .sink(input, false);
              env.execute();
          }
      }

Menulis data dari Flink yang sepenuhnya dikelola

1. Prasyarat: Buat tabel MaxCompute.

   Anda harus membuat tabel MaxCompute tujuan untuk data Flink. Contoh berikut menunjukkan cara membuat tabel delta.

   SET odps.sql.type.system.odps2=true;
   DROP TABLE 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") ;

2. Konektor Flink telah dipra-instal pada Flink yang sepenuhnya dikelola, sehingga tidak diperlukan instalasi manual. Anda dapat melihat detail konektor di Konsol Realtime Compute for Apache Flink.

3. Buat tabel Flink, buat data Flink real-time menggunakan pekerjaan Flink SQL, lalu deploy pekerjaan tersebut setelah pengembangan.

   Di halaman pengembangan pekerjaan Flink, buat dan edit pekerjaan SQL. Contoh berikut mendefinisikan tabel sumber yang menghasilkan data acak, tabel hasil yang terhubung ke MaxCompute, dan pernyataan INSERT untuk mentransfer data. Untuk informasi lebih lanjut tentang cara mengembangkan pekerjaan SQL, lihat Peta pengembangan pekerjaan.

   -- Buat tabel sumber Flink.
   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''}'
   );
   
   -- Buat tabel hasil sementara di Flink.
   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'
   );
   
   -- Logika komputasi Flink.
   INSERT INTO test_c_d_g
   SELECT  c1 AS c1,
           c2 AS c2,
           gt AS gt,
           date_format(gt, 'yyyyMMddHH') AS ds
   FROM    fake_src_table;

   Parameter:

   odps.end.point: Gunakan Titik akhir jaringan internal dari wilayah yang sesuai.

   upsert.write.bucket.num: Nilai ini harus konsisten dengan nilai properti write.bucket.num dari tabel delta yang dibuat di MaxCompute.

4. Kueri tabel MaxCompute untuk memverifikasi bahwa data telah ditulis.

   SELECT * FROM mf_flink_upsert WHERE ds=2023061517;
   
   -- Hasil Anda mungkin berbeda karena data sumber dihasilkan secara acak.
   +------+----+------+----+
   | 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 |
   

Lampiran: Parameter konektor Flink

• Parameter dasar

  Parameter	Wajib	Nilai default	Deskripsi
  connector	Ya	—	Atur jenis konektor ke MaxCompute.
  odps.project.name	Ya	—	Nama proyek MaxCompute.
  odps.access.id	Ya	—	ID AccessKey akun Anda. Lihat halaman Pasangan AccessKey.
  odps.access.key	Ya	—	Rahasia AccessKey akun Anda. Lihat halaman Pasangan AccessKey.
  odps.end.point	Ya	—	Titik akhir MaxCompute. Untuk daftar Titik akhir berdasarkan wilayah, lihat Titik akhir.
  odps.tunnel.end.point	Tidak	—	Titik akhir publik layanan Tunnel. Secara default, permintaan akan diarahkan secara otomatis ke Titik akhir Tunnel yang sesuai. Atur parameter ini untuk menggunakan Titik akhir tertentu dan menonaktifkan pengarahan otomatis. Untuk informasi lebih lanjut tentang Titik akhir Tunnel di berbagai wilayah dan jaringan, lihat Titik akhir.
  odps.tunnel.quota.name	Tidak	—	Nama kuota Tunnel yang digunakan untuk mengakses MaxCompute.
  table.name	Ya	—	Nama tabel MaxCompute dalam format [project.][schema.]table.
  odps.namespace.schema	Tidak	false	Menentukan apakah akan menggunakan model tiga lapis. Untuk informasi lebih lanjut tentang model tiga lapis, lihat Operasi skema.
  sink.operation	Ya	insert	Jenis penulisan. Nilai yang valid adalah insert atau upsert. Catatan Mode upsert hanya didukung untuk tabel delta MaxCompute.
  sink.parallelism	Tidak	—	Paralelisme penulisan. Jika tidak diatur, nilai ini akan menggunakan paralelisme sumber hulu secara default. Catatan Pastikan properti tabel write.bucket.num merupakan kelipatan bilangan bulat dari nilai konfigurasi ini untuk kinerja penulisan optimal dan penghematan memori maksimal pada node Sink.
  sink.meta.cache.time	Tidak	400	Ukuran cache metadata.
  sink.meta.cache.expire.time	Tidak	1200	Waktu kedaluwarsa cache untuk metadata, dalam detik.
  sink.coordinator.enable	Tidak	true	Menentukan apakah akan mengaktifkan mode koordinator.

• Parameter partisi

  Parameter	Wajib	Nilai default	Deskripsi
  sink.partition	Tidak	—	Nama partisi tempat data ditulis. Jika Anda menggunakan partisi dinamis, parameter ini menentukan nama partisi induk dari partisi dinamis.
  sink.partition.default-value	Tidak	__DEFAULT_PARTITION__	Nama partisi default saat menggunakan partisi dinamis.
  sink.dynamic-partition.limit	Tidak	100	Jumlah maksimum partisi yang dapat ditulis secara bersamaan dalam satu checkpoint selama partisi dinamis. Catatan Menambah nilai ini secara signifikan dapat menyebabkan error kehabisan memori (OOM) pada node sink. Jika jumlah partisi bersamaan melebihi batas ini, pekerjaan akan gagal.
  sink.group-partition.enable	Tidak	false	Menentukan apakah akan mengelompokkan berdasarkan partisi saat menggunakan partisi dinamis.
  sink.partition.assigner.class	Tidak	—	Kelas implementasi PartitionAssigner.

• Parameter penulisan mode FileCached

  Gunakan mode cache file untuk pekerjaan dengan jumlah partisi dinamis yang besar. Parameter berikut mengonfigurasi mode ini.

  Parameter	Wajib	Nilai default	Deskripsi
  sink.file-cached.enable	Tidak	false	Mengaktifkan mode FileCached. Direkomendasikan untuk pekerjaan dengan jumlah partisi dinamis yang besar. false: Mode FileCached dinonaktifkan. true: Mode FileCached diaktifkan. Catatan Saat jumlah partisi dinamis besar, Anda dapat menggunakan mode cache file.
  sink.file-cached.tmp.dirs	Tidak	./local	Direktori cache file default dalam mode FileCached.
  sink.file-cached.writer.num	Tidak	16	Jumlah thread unggah data konkuren untuk satu tugas dalam mode FileCached. Catatan Jangan menaikkan nilai parameter ini secara signifikan. Jika jumlah partisi yang ditulis secara bersamaan berlebihan, kemungkinan besar akan terjadi error OOM.
  sink.bucket.check-interval	Tidak	60000	Interval pemeriksaan ukuran file dalam mode FileCached. Satuan: milidetik (ms).
  sink.file-cached.rolling.max-size	Tidak	16 M	Ukuran maksimum untuk satu file cache. Saat file melebihi ukuran ini, file tersebut akan diunggah.
  sink.file-cached.memory	Tidak	64 M	Ukuran maksimum memori off-heap yang digunakan untuk penulisan file dalam mode FileCached.
  sink.file-cached.memory.segment-size	Tidak	128 KB	Ukuran buffer yang digunakan untuk penulisan file dalam mode FileCached.
  sink.file-cached.flush.always	Tidak	true	Menentukan apakah akan menggunakan cache untuk penulisan file dalam mode FileCached.
  sink.file-cached.write.max-retries	Tidak	3	Jumlah percobaan ulang untuk mengunggah data dalam mode FileCached.

• Parameter penulisan Insert atau Upsert

  Parameter penulisan Upsert

  Parameter	Wajib	Nilai default	Deskripsi
  upsert.writer.max-retries	Tidak	3	Jumlah percobaan ulang setelah penulis upsert gagal menulis data ke bucket.
  upsert.writer.buffer-size	Tidak	64 MB	Ukuran cache untuk satu penulis upsert di Flink. Catatan Saat jumlah ukuran buffer semua bucket mencapai ambang batas yang telah ditentukan, sistem secara otomatis memicu operasi flush untuk memperbarui data ke server. Penulis upsert menulis data ke beberapa bucket secara bersamaan. Kami merekomendasikan agar Anda menaikkan nilai parameter ini untuk meningkatkan efisiensi penulisan. Jika data ditulis ke sejumlah besar partisi, kemungkinan terjadi error OOM. Dalam kasus ini, Anda dapat mengurangi nilai parameter ini.
  upsert.writer.bucket.buffer-size	Tidak	1 MB	Ukuran cache untuk satu bucket di Flink. Jika sumber daya memori pada server Flink tidak mencukupi, Anda dapat mengurangi nilai parameter ini.
  upsert.write.bucket.num	Ya	—	Jumlah bucket untuk tabel tujuan harus sama dengan nilai write.bucket.num.
  upsert.write.slot-num	Tidak	1	Jumlah slot Tunnel yang digunakan oleh satu sesi.
  upsert.commit.max-retries	Tidak	3	Jumlah percobaan ulang untuk commit sesi upsert.
  upsert.commit.thread-num	Tidak	16	Paralelisme commit sesi upsert. Jangan mengatur nilai ini terlalu tinggi. Jumlah commit konkuren yang tinggi menyebabkan peningkatan konsumsi sumber daya, yang dapat menyebabkan masalah kinerja atau konsumsi sumber daya berlebihan.
  upsert.major-compact.min-commits	Tidak	100	Jumlah minimum commit yang diperlukan untuk memicu kompaksi mayor.
  upsert.commit.timeout	Tidak	600	Periode timeout untuk commit sesi upsert. Satuan: detik (s).
  upsert.major-compact.enable	Tidak	false	Menentukan apakah akan mengaktifkan kompaksi mayor.
  upsert.flush.concurrent	Tidak	2	Jumlah maksimum bucket tempat data dapat ditulis secara bersamaan dalam satu partisi. Catatan Saat data dalam bucket di-flush, slot Tunnel akan digunakan.

  Catatan

  Untuk informasi lebih lanjut tentang konfigurasi parameter yang direkomendasikan untuk penulisan upsert, lihat Konfigurasi parameter yang direkomendasikan untuk penulisan upsert.

  Parameter penulisan Insert

  Parameter	Wajib	Nilai default	Deskripsi
  insert.commit.thread-num	Tidak	16	Paralelisme sesi commit.
  insert.arrow-writer.enable	Tidak	false	Menentukan apakah akan menggunakan format Arrow.
  insert.arrow-writer.batch-size	Tidak	512	Jumlah maksimum baris dalam satu batch Arrow.
  insert.arrow-writer.flush-interval	Tidak	100000	Interval flush penulis. Satuan: milidetik (ms).
  insert.writer.buffer-size	Tidak	64 MB	Ukuran cache penulis ter-buffer.