DataWorks:Sumber data Hologres

Baca dan tulis batch

• Hologres Writer tidak mendukung penulisan data ke tabel eksternal Hologres.

• Logika untuk memperoleh endpoint Hologres guna koneksi sumber data:

  • Untuk instans Hologres di wilayah saat ini, prioritas endpoint adalah: any Tunnel > Single Tunnel > Public (internet).

  • Untuk instans Hologres di wilayah lain, prioritas endpoint adalah: Public (internet) > Single Tunnel.

Baca real-time tabel tunggal

• Versi Hologres harus 2.1 atau lebih baru.

• Sinkronisasi inkremental untuk tabel partisi Hologres tidak didukung.

• Sinkronisasi pesan perubahan DDL pada tabel Hologres tidak didukung.

• Sinkronisasi inkremental Hologres mendukung tipe data berikut:

  INTEGER, BIGINT, TEXT, CHAR(n), VARCHAR(n), REAL, JSON, SERIAL, OID, INT4[], INT8[], FLOAT8[], BOOLEAN[], TEXT[].

• Untuk sinkronisasi real-time tabel tunggal dari Hologres, Anda harus mengaktifkan Hologres Binlog pada tabel sumber Hologres. Untuk informasi selengkapnya, lihat Berlangganan Hologres Binlog.

Tulis real-time basis data penuh

• Tugas sinkronisasi data real-time tidak mendukung sinkronisasi tabel tanpa primary key.

• Saat menyinkronkan data dari basis data penuh MySQL ke Hologres secara real time, data hanya dapat ditulis ke tabel anak dari tabel partisi. Penulisan ke tabel induk tabel partisi tidak didukung.

Baca batch

Hologres Reader mendukung dua mode baca:

Mode JDBC (default)

Membaca data dari tabel Hologres melalui PSQL. Beberapa tugas konkuren diinisiasi berdasarkan jumlah shard tabel, dengan setiap shard sesuai dengan satu tugas Select konkuren:

• Saat membuat tabel di Hologres, Anda dapat mengonfigurasi jumlah shard tabel dengan menjalankan CALL set_table_property('table_name', 'shard_count', 'xx') dalam transaksi CREATE TABLE yang sama.

  Secara default, database menggunakan jumlah shard default, yang bergantung pada konfigurasi instans Hologres.

• Pernyataan Select memfilter data berdasarkan shard menggunakan kolom bawaan hg_shard_id dari tabel.

Mode Arrow (useArrow=true)

Membaca data melalui protokol Hologres COPY OUT Arrow. Mode ini mendukung transfer terkompresi LZ4 dan memberikan kinerja lebih tinggi:

• Memerlukan Hologres versi >= 4.0.18.

• Tugas konkuren juga dibagi berdasarkan jumlah shard.

• Kolom fungsi dan kolom konstan tidak didukung. Hanya kolom fisik dalam tabel yang didukung.

• Transfer terkompresi diaktifkan secara otomatis ketika persyaratan versi terpenuhi.

Tulis batch

Hologres Writer memperoleh data protokol yang dihasilkan oleh Reader melalui framework sinkronisasi data. Saluran dan strategi resolusi konflik yang digunakan untuk menulis data ditentukan oleh konfigurasi writeMode dan conflictMode.

Mode tulis (writeMode)

Rekomendasi skenario

Mode penanganan konflik (conflictMode)

Anda dapat mengonfigurasi conflictMode untuk menentukan cara menangani data yang baru diimpor ketika terjadi konflik primary key dengan data yang sudah ada:

conflictMode hanya berlaku untuk tabel yang memiliki primary key. Untuk informasi lebih lanjut tentang mekanisme penulisan dan kinerja, lihat Menulis data ke Hologres.

• Ketika conflictMode diatur ke Replace (full row update), data baru menimpa data lama. Semua kolom dalam baris tersebut ditimpa. Kolom yang tidak termasuk dalam pemetaan kolom dipaksakan menjadi NULL.

• Ketika conflictMode diatur ke Update, data baru menimpa data lama. Hanya kolom yang termasuk dalam pemetaan kolom yang ditimpa.

• Ketika conflictMode diatur ke Ignore, data baru diabaikan.

Sinkronisasi batch tabel tunggal

• Sumber data yang didukung: Semua jenis sumber data yang didukung oleh modul Data Integration.

• Panduan konfigurasi: Konfigurasikan tugas sinkronisasi batch di antarmuka tanpa kode dan Konfigurasikan tugas sinkronisasi batch menggunakan editor kode. Untuk parameter lengkap dan demo skrip yang tersedia di editor kode, lihat Lampiran: Parameter lengkap dan demo skrip di bawah ini.

Sinkronisasi real-time tabel tunggal

• Sumber data yang didukung: DataHub, Hologres, Kafka, dan LogHub

• Panduan konfigurasi: Konfigurasikan tugas sinkronisasi real-time.

Sinkronisasi batch basis data penuh

• Sumber data yang didukung: AnalyticDB for MySQL 3.0, ClickHouse, Doris, Hologres, Oracle, PolarDB, dan SQL Server

• Panduan konfigurasi: Konfigurasikan tugas sinkronisasi batch basis data penuh

Sinkronisasi real-time basis data penuh

• Sumber data yang didukung: AnalyticDB for OceanBase, MongoDB, MySQL, Oracle, PolarDB, PolarDB-X 2.0, dan PostgreSQL

• Panduan konfigurasi: Konfigurasikan tugas sinkronisasi real-time untuk menyinkronkan data satu tabel

Real-time basis data penuh Serverless

• Sumber data yang didukung: MySQL

• Panduan konfigurasi: Konfigurasikan tugas sinkronisasi real-time basis data penuh Serverless

Konfigurasikan tugas sinkronisasi batch menggunakan editor kode

Jika ingin mengonfigurasi tugas sinkronisasi batch menggunakan editor kode, Anda harus mengonfigurasi parameter terkait dalam skrip berdasarkan persyaratan format skrip terpadu. Untuk informasi selengkapnya, lihat Konfigurasi mode skrip. Informasi berikut menjelaskan parameter yang harus dikonfigurasi untuk sumber data saat mengonfigurasi tugas sinkronisasi batch menggunakan editor kode.

Demo skrip Reader

• Konfigurasikan tabel non-partisi

  • Contoh berikut menunjukkan cara mengonfigurasi pembacaan data dari tabel non-partisi Hologres ke memori.

    {
        "transform": false,
        "type": "job",
        "version": "2.0",
        "steps": [
            {
                "stepType": "holo",
                "parameter": {
                    "datasource": "holo_db",
                    "envType": 1,
                    "column": [ 
                        "tag",
                        "id",
                        "title",
                        "body"
                    ],
                    "where": "",
                    "table": "holo_reader_basic_src"
                },
                "name": "Reader",
                "category": "reader"
            },
            {
                "stepType": "stream",
                "parameter": {
                    "print": false,
                    "fieldDelimiter": ","
                },
                "name": "Writer",
                "category": "writer"
            }
        ],
        "setting": {
            "executeMode": null,
            "failoverEnable": null,
            "errorLimit": {
                "record": "0"
            },
            "speed": {
                "concurrent": 2,
                "throttle": false
            }
        },
        "order": {
            "hops": [
                {
                    "from": "Reader",
                    "to": "Writer"
                }
            ]
        }
    }

  • Pernyataan DDL tabel Hologres adalah sebagai berikut.

    begin;
    drop table if exists holo_reader_basic_src;
    create table holo_reader_basic_src(
      tag text not null, 
      id int not null, 
      title text not null, 
      body text, 
      primary key (tag, id));
      call set_table_property('holo_reader_basic_src', 'orientation', 'column');
      call set_table_property('holo_reader_basic_src', 'shard_count', '3');
    commit;

• Konfigurasikan tabel partisi

  • Konfigurasikan pembacaan data dari tabel anak tabel partisi Hologres ke memori.

    Catatan

    Perhatikan konfigurasi partition.

    {
        "transform": false,
        "type": "job",
        "version": "2.0",
        "steps": [
            {
                "stepType": "holo",
                "parameter": {
                    "selectedDatabase": "public",
                    "partition": "tag=foo",
                    "datasource": "holo_db",
                    "envType": 1,
                    "column": [
                        "tag",
                        "id",
                        "title",
                        "body"
                    ],
                    "tableComment": "",
                    "where": "",
                    "table": "public.holo_reader_basic_part_src"
                },
                "name": "Reader",
                "category": "reader"
            },
            {
          "stepType":"stream",
          "parameter":{},
          "name":"Writer",
          "category":"writer"
        }
      ],
      "setting":{
        "errorLimit":{
          "record":"0"
        },
        "speed":{
          "throttle":true,
          "concurrent":1,
          "mbps":"12"
            }
         },
      "order":{
        "hops":[
          {
            "from":"Reader",
            "to":"Writer"
          }
        ]
      }
    }

  • Pernyataan DDL tabel Hologres adalah sebagai berikut.

    begin;
    drop table if exists holo_reader_basic_part_src;
    create table holo_reader_basic_part_src(
      tag text not null, 
      id int not null, 
      title text not null, 
      body text, 
      primary key (tag, id))
      partition by list( tag );
      call set_table_property('holo_reader_basic_part_src', 'orientation', 'column');
      call set_table_property('holo_reader_basic_part_src', 'shard_count', '3');
    commit;
    
    create table holo_reader_basic_part_src_1583161774228 partition of holo_reader_basic_part_src for values in ('foo');
    
    # Pastikan tabel anak dari tabel partisi telah dibuat dan data telah diimpor.
    postgres=# \d+ holo_reader_basic_part_src
                             Table "public.holo_reader_basic_part_src"
     Column |  Type   | Collation | Nullable | Default | Storage  | Stats target | Description 
    --------+---------+-----------+----------+---------+----------+--------------+-------------
     tag    | text    |           | not null |         | extended |              | 
     id     | integer |           | not null |         | plain    |              | 
     title  | text    |           | not null |         | extended |              | 
     body   | text    |           |          |         | extended |              | 
    Partition key: LIST (tag)
    Indexes:
        "holo_reader_basic_part_src_pkey" PRIMARY KEY, btree (tag, id)
    Partitions: holo_reader_basic_part_src_1583161774228 FOR VALUES IN ('foo')

Parameter skrip Reader

Demo skrip Writer

• Konfigurasikan tabel non-partisi

  • Konfigurasikan data yang dihasilkan dari MySQL untuk diimpor ke tabel Hologres biasa. Contoh berikut menunjukkan konfigurasi untuk mengimpor data dalam mode INSERT.

    {
        "type": "job",
        "version": "2.0",
        "steps": [
            {
                "stepType": "mysql",
                "parameter": {
                    "envType": 0,
                    "useSpecialSecret": false,
                    "column": [
                        "<column1>",
                        "<column2>",
                        ......,
                        "<columnN>"
                    ],
                    "tableComment": "",
                    "connection": [
                        {
                            "datasource": "<mysql_source_name>",//Nama sumber data MySQL
                            "table": [
                                "<mysql_table_name>"
                            ]
                        }
                    ],
                    "where": "",
                    "splitPk": "",
                    "encoding": "UTF-8"
                },
                "name": "Reader",
                "category": "reader"
            },
            {
                "stepType": "holo",
                "parameter": {
                    "selectedDatabase":"public",
                    "schema": "public",
                    "writeMode": "FIXED_COPY",
                    "maxConnectionCount": 9,
                    "truncate":true,//Aturan pembersihan
                    "datasource": "<holo_sink_name>",//Nama sumber data Hologres
                    "conflictMode": "ignore",
                    "envType": 0,
                    "column": [
                        "<column1>",
                        "<column2>",
                        ......,
                        "<columnN>"
                    ],
                    "tableComment": "",
                    "table": "<holo_table_name>",
                    "reShuffleByDistributionKey":false
                },
                "name": "Writer",
                "category": "writer"
            }
        ],
        "setting": {
            "executeMode": null,
            "errorLimit": {
                "record": "0"
            },
            "locale": "zh_CN",
            "speed": {
                "concurrent": 2,//Konkurensi pekerjaan
                "throttle": false//Pengendalian aliran
            }
        },
        "order": {
            "hops": [
                {
                    "from": "Reader",
                    "to": "Writer"
                }
            ]
        }
    }

  • Pernyataan DDL untuk tabel Hologres adalah sebagai berikut.

    begin;
    drop table if exists mysql_to_holo_test;
    create table mysql_to_holo_test(
      tag text not null,
      id int not null,
      body text not null,
      brrth date,
      primary key (tag, id));
      call set_table_property('mysql_to_holo_test', 'orientation', 'column');
      call set_table_property('mysql_to_holo_test', 'distribution_key', 'id');
      call set_table_property('mysql_to_holo_test', 'clustering_key', 'birth');
    commit;

• Konfigurasikan tabel partisi

  Catatan

  • Hologres hanya mendukung partisi LIST. Hanya satu kolom partisi yang didukung, dan kolom tersebut harus bertipe INT4 atau TEXT.

  • Pastikan parameter ini sesuai dengan konfigurasi partisi dalam DDL tabel.

  • Konfigurasikan sinkronisasi data yang dihasilkan dari MySQL ke tabel anak tabel partisi Hologres.

    {
      "type": "job",
      "version": "2.0",
      "steps": [
        {
          "stepType": "mysql",
          "parameter": {
            "envType": 0,
            "useSpecialSecret": false,
            "column": [
              "<column1>",
              "<column2>",
                ......,
              "<columnN>"
            ],
            "tableComment": "",
            "connection": [
              {
                "datasource": "<mysql_source_name>",
                "table": [
                  "<mysql_table_name>"
                ]
              }
            ],
            "where": "",
            "splitPk": "<mysql_pk>",//Bidang primary key MySQL
            "encoding": "UTF-8"
          },
          "name": "Reader",
          "category": "reader"
        },
        {
          "stepType": "holo",
          "parameter": {
            "selectedDatabase": "public",
            "writeMode": "insert",
            "maxConnectionCount": 9,
            "partition": "ds=20201215",//Kunci partisi Hologres
            "truncate": "false",
            "datasource": "<holo_sink_name>",//Nama sumber data Hologres
            "conflictMode": "ignore",
            "envType": 0,
            "column": [
              "<column1>",
              "<column2>",
                ......,
              "<columnN>"
            ],
            "tableComment": "",
            "table": "<holo_table_name>",
            "reShuffleByDistributionKey":false
          },
          "name": "Writer",
          "category": "writer"
        }
      ],
      "setting": {
        "executeMode": null,
        "failoverEnable": null,
        "errorLimit": {
          "record": "0"
        },
        "speed": {
          "concurrent": 2,//Konkurensi pekerjaan
          "throttle": false//Pengendalian aliran
        }
      },
      "order": {
        "hops": [
          {
            "from": "Reader",
            "to": "Writer"
          }
        ]
      }
    }

  • Pernyataan DDL tabel Hologres adalah sebagai berikut.

    BEGIN;
    CREATE TABLE public.hologres_parent_table(
      a text ,
      b int,
      c timestamp,
      d text,
      ds text,
      primary key(ds,b)
      )
      PARTITION BY LIST(ds);
    CALL set_table_property('public.hologres_parent_table', 'orientation', 'column');
    CREATE TABLE public.holo_child_1 PARTITION OF public.hologres_parent_table FOR VALUES IN('20201215');
    CREATE TABLE public.holo_child_2 PARTITION OF public.hologres_parent_table FOR VALUES IN('20201216');
    CREATE TABLE public.holo_child_3 PARTITION OF public.hologres_parent_table FOR VALUES IN('20201217');
    COMMIT;

Parameter skrip Writer

Parameter dasar

Parameter koneksi dan kinerja

Parameter khusus mode FIXED_COPY

Parameter khusus mode STAGE