Cara menyinkronkan data Doris dan bidang yang didukung - DataWorks

Sumber data Doris memungkinkan Anda membaca dan menulis data ke database Doris untuk pemrosesan data skala besar. Topik ini menjelaskan cara menggunakan DataWorks untuk menyinkronkan data dengan Doris.

Tipe data yang didukung

Versi Doris yang berbeda mendukung tipe data dan model agregasi yang berbeda. Untuk informasi lengkap mengenai semua tipe data yang didukung di setiap versi Doris, lihat dokumentasi resmi Doris. Tabel berikut menjelaskan tipe data utama yang didukung.

Data type	Supported model	Doris version
SMALLINT	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
INT	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
BIGINT	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
LARGEINT	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
FLOAT	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
DOUBLE	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
DECIMAL	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
DECIMALV3	Aggregate,Unique,Duplicate	Versions later than 1.2.1, 2.x
DATE	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
DATETIME	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
DATEV2	Aggregate,Unique,Duplicate	1.2.x, 2.x
DATATIMEV2	Aggregate,Unique,Duplicate	1.2.x, 2.x
CHAR	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
VARCHAR	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
STRING	Aggregate,Unique,Duplicate	0.x.x, 1.1.x, 1.2.x, 2.x
VARCHAR	Aggregate,Unique,Duplicate	1.1.x, 1.2.x, 2.x
ARRAY	Duplicate	1.2.x, 2.x
JSONB	Aggregate,Unique,Duplicate	1.2.x, 2.x
HLL	Aggregate	0.x.x, 1.1.x, 1.2.x, 2.x
BITMAP	Aggregate	0.x.x, 1.1.x, 1.2.x, 2.x
QUANTILE_STATE	Aggregate	1.2.x, 2.x

Persiapkan lingkungan ApsaraDB for OceanBase sebelum sinkronisasi data

Sebelum menggunakan DataWorks untuk menyinkronkan data ke sumber data Doris, Anda harus terlebih dahulu menyiapkan lingkungan Doris. Hal ini memastikan bahwa tugas sinkronisasi data dapat dikonfigurasi dan berjalan sesuai harapan. Berikut ini adalah langkah-langkah untuk menyiapkan lingkungan Doris guna sinkronisasi data.

Buat akun dan berikan izin

Anda harus membuat akun untuk login ke database Doris guna operasi selanjutnya dan menentukan kata sandi untuk akun tersebut agar dapat terhubung ke database. Jika Anda ingin menggunakan user root bawaan Doris, Anda harus menetapkan kata sandi karena secara default user root tidak memiliki kata sandi. Anda dapat menjalankan pernyataan SQL berikut di Doris untuk menetapkan kata sandi:

SET PASSWORD FOR 'root' = PASSWORD('Password')

Konfigurasi koneksi jaringan untuk Doris

Untuk menulis data menggunakan metode StreamLoad, Anda perlu mengakses alamat IP pribadi node FE. Jika Anda mengakses alamat IP publik node FE, permintaan akan dialihkan ke alamat IP pribadi node BE. Untuk informasi lebih lanjut mengenai pengalihan ini, lihat Data operation issues. Dalam kasus ini, Anda harus membuat koneksi jaringan antara sumber data Anda dan serverless resource group atau grup sumber daya eksklusif untuk Integrasi Data agar grup sumber daya tersebut dapat mengakses sumber data melalui jaringan internal. Untuk informasi lebih lanjut tentang cara membuat koneksi jaringan antara database Doris dan grup sumber daya, lihat Network connectivity solutions.

Tambahkan sumber data

Sebelum mengembangkan tugas sinkronisasi di DataWorks, Anda harus menambahkan sumber data yang diperlukan ke DataWorks dengan mengikuti petunjuk dalam Data source management. Anda dapat melihat deskripsi parameter di Konsol DataWorks untuk memahami arti parameter saat menambahkan sumber data.

Perhatikan persyaratan konfigurasi untuk item konfigurasi berikut dari sumber data Doris:

JdbcUrl: Masukkan string koneksi JDBC yang mencakup alamat IP, nomor port, nama database, dan parameter koneksi. Alamat IP publik maupun pribadi didukung. Jika Anda menggunakan alamat IP publik, pastikan grup sumber daya Integrasi Data dapat mengakses host tempat instans Doris berada.
FE endpoint: Masukkan alamat IP dan port node FE. Jika kluster Anda memiliki beberapa node FE, Anda dapat memasukkan beberapa endpoint yang dipisahkan koma, misalnya, ip1:port1,ip2:port2. Saat menguji koneksi, DataWorks menguji konektivitas ke semua FE endpoints yang ditentukan.
Username: Masukkan username untuk mengakses database Doris.
Password: Masukkan kata sandi yang sesuai dengan username tersebut.

Kembangkan tugas sinkronisasi data

Untuk informasi mengenai titik masuk dan prosedur konfigurasi tugas sinkronisasi, lihat panduan konfigurasi berikut.

Untuk informasi selengkapnya tentang prosedur konfigurasi, lihat Mengonfigurasi tugas sinkronisasi Batch menggunakan Antarmuka tanpa kode dan Mengonfigurasi tugas sinkronisasi Batch menggunakan editor kode.
Untuk informasi mengenai semua parameter yang dikonfigurasi dan kode yang dijalankan saat Anda menggunakan editor kode untuk mengonfigurasi tugas sinkronisasi batch, lihat Appendix: Code and parameters.

Lampiran: Kode dan parameter

Konfigurasi tugas sinkronisasi batch menggunakan editor kode

Jika Anda ingin mengonfigurasi tugas sinkronisasi batch menggunakan editor kode, Anda harus mengonfigurasi parameter terkait dalam skrip sesuai dengan format skrip terpadu. Untuk informasi lebih lanjut, lihat Use the Code Editor. Informasi berikut menjelaskan parameter yang harus Anda konfigurasi untuk sumber data saat menggunakan editor kode untuk tugas sinkronisasi batch.

Contoh skrip Reader

{
  "type": "job",
  "version": "2.0",// Nomor versi. 
  "steps": [
    {
      "stepType": "doris",// Nama plug-in. 
      "parameter": {
        "column": [// Nama kolom. 
          "id"
        ],
        "connection": [
          {
            "querySql": [
              "select a,b from join1 c join join2 d on c.id = d.id;"
            ],
            "datasource": ""// Nama sumber data. 
          }
        ],
        "where": "",// Klausul WHERE. 
        "splitPk": "",// Kunci shard. 
        "encoding": "UTF-8"// Format encoding. 
      },
      "name": "Reader",
      "category": "reader"
    },
    {
      "stepType": "stream",
      "parameter": {},
      "name": "Writer",
      "category": "writer"
    }
  ],
  "setting": {
    "errorLimit": {
      "record": "0"// Jumlah maksimum catatan data kotor yang diizinkan. 
    },
    "speed": {
      "throttle": true,// Menentukan apakah pembatasan kecepatan diaktifkan. Nilai false berarti pembatasan kecepatan dinonaktifkan, dan nilai true berarti pembatasan kecepatan diaktifkan. Parameter mbps hanya berlaku ketika parameter throttle diatur ke true. 
      "concurrent": 1,// Jumlah maksimum thread paralel. 
      "mbps": "12"// Laju transmisi maksimum. Satuan: MB/s. 
    }
  },
  "order": {
    "hops": [
      {
        "from": "Reader",
        "to": "Writer"
      }
    ]
  }
}

Parameter skrip Reader

Parameter	Deskripsi	Wajib	Nilai default
datasource	Nama sumber data. Harus sama dengan nama sumber data yang telah ditambahkan. Anda dapat menambahkan sumber data menggunakan editor kode.	Ya	Tidak ada nilai default
table	Nama tabel dari mana Anda ingin membaca data. Setiap tugas sinkronisasi hanya dapat digunakan untuk menyinkronkan data dari satu tabel. Untuk tabel yang di-shard, Anda dapat menggunakan parameter table untuk menentukan partisi dari mana Anda ingin membaca data. Contoh: Anda dapat mengonfigurasi rentang untuk membaca data dari database dan tabel yang di-shard. Misalnya, `'table_[0-99]'` menentukan bahwa data dibaca dari `'table_0'`, `'table_1'`, dan `'table_2'`, hingga `'table_99'`. Jika tabel Anda memiliki akhiran numerik dengan panjang yang sama, seperti `'table_000'`, `'table_001'`, dan `'table_002'`, hingga `'table_999'`, Anda dapat mengonfigurasi parameter sebagai `'"table": ["table_00[0-9]", "table_0[10-99]", "table_[100-999]"]'`. Catatan Doris Reader membaca data dari kolom yang ditentukan oleh parameter column dalam partisi yang ditentukan oleh parameter table. Jika partisi atau kolom yang ditentukan tidak ada, tugas sinkronisasi akan gagal.	Ya	Tidak ada nilai default
column	Kolom yang ingin Anda sinkronkan. Kolom dijelaskan dalam array JSON. Secara default, semua kolom disinkronkan. Misalnya, gunakan `[""]`. Pemangkasan kolom: Anda dapat memilih subset kolom untuk diekspor. Urutan kolom dapat diubah. Artinya, Anda dapat menentukan kolom dalam urutan yang berbeda dari urutan yang ditentukan oleh skema tabel sumber. Konfigurasi konstanta: Anda harus mengikuti sintaks SQL Doris. Contoh: ["id","`table`","1","'mingya.wmy'","'null'","to_char(a+1)","2.3","true"]. id: nama kolom. table: nama kolom yang berisi kata kunci yang dicadangkan. 1: konstanta integer. 'mingya.wmy': konstanta string, yang diapit tanda kutip tunggal ('). null: `" "` menunjukkan string kosong. `null` menunjukkan nilai null. `'null'` menunjukkan string "null". to_char(a+1): ekspresi fungsi yang digunakan untuk menghitung panjang string. 2.3: konstanta floating-point. true: Nilai Boolean. Parameter column* harus secara eksplisit menentukan semua kolom dari mana Anda ingin membaca data. Parameter ini tidak boleh dibiarkan kosong.	Ya	Tidak ada nilai default
splitPk	Untuk meningkatkan performa baca, Anda dapat menggunakan parameter splitPk untuk menentukan shard key. Integrasi Data menggunakan kunci ini untuk mempartisi data dan menjalankan tugas secara konkuren. Kami menyarankan Anda mengatur parameter splitPk ke nama kolom kunci primer tabel. Data dapat didistribusikan secara merata ke shard yang berbeda berdasarkan kolom kunci primer, bukan hanya terkonsentrasi pada shard tertentu. Parameter splitPk hanya mendukung sharding untuk data bertipe integer. Jika Anda mengatur parameter splitPk ke field dengan tipe data yang tidak didukung, seperti string, floating point, atau tanggal, pengaturan parameter ini akan diabaikan, dan satu thread saja yang digunakan untuk membaca data. Jika parameter splitPk tidak diberikan atau dibiarkan kosong, satu thread saja yang digunakan untuk membaca data.	Tidak	Tidak ada nilai default
where	Kondisi filter. Dalam banyak skenario bisnis, Anda mungkin hanya ingin menyinkronkan data hari ini. Untuk melakukan ini, Anda dapat menentukan kondisi where sebagai `gmt_create > $bizdate`. Anda dapat menggunakan klausa WHERE untuk membaca data inkremental. Jika parameter where tidak diberikan atau dibiarkan kosong, Doris Reader membaca semua data. Jangan mengatur parameter where menjadi limit 10. Nilai ini tidak sesuai dengan batasan Doris pada klausa SQL WHERE.	Tidak	Tidak ada nilai default
querySql (parameter advanced, hanya tersedia di editor kode)	Dalam beberapa skenario bisnis, parameter where mungkin tidak cukup untuk menggambarkan kondisi filter yang diinginkan. Anda dapat menggunakan parameter ini untuk menentukan kueri SQL kustom untuk filtering. Jika Anda mengonfigurasi parameter ini, Integrasi Data akan mengabaikan parameter table, column, where, dan splitPk, serta menggunakan kueri kustom untuk mengambil data. Misalnya, untuk melakukan join beberapa tabel sebelum sinkronisasi, Anda dapat menggunakan kueri seperti `select a,b from table_a join table_b on table_a.id = table_b.id`. Parameter querySql memiliki prioritas lebih tinggi daripada parameter table, column, where, dan splitPk. Parameter datasource tetap diperlukan agar sistem dapat mengambil detail koneksi seperti username dan password. Catatan Nama parameter querySql bersifat case-sensitive. Misalnya, querysql tidak berlaku.	Tidak	Tidak ada nilai default

Contoh skrip Writer

{
  "stepType": "doris",// Nama plug-in. 
  "parameter":
  {
    "postSql":// Pernyataan SQL yang ingin dijalankan setelah tugas sinkronisasi dijalankan. 
    [],
    "preSql":
    [],// Pernyataan SQL yang ingin dijalankan sebelum tugas sinkronisasi dijalankan. 
    "datasource":"doris_datasource",// Nama sumber data. 
    "table": "doris_table_name",// Nama tabel. 
    "column":
    [
      "id",
      "table_id",
      "table_no",
      "table_name",
      "table_status"
    ],
    "loadProps":{
      "column_separator": "\\x01",// Pemisah kolom data dalam format CSV.
      "line_delimiter": "\\x02"// Pemisah baris data dalam format CSV.
    }
  },
  "name": "Writer",
  "category": "writer"
}

Parameter skrip Writer

Parameter	Deskripsi	Wajib	Nilai default
datasource	Nama sumber data. Harus sama dengan nama sumber data yang telah ditambahkan. Anda dapat menambahkan sumber data menggunakan editor kode.	Ya	Tidak ada nilai default
table	Nama tabel dari mana Anda ingin membaca data.	Ya	Tidak ada nilai default
column	Kolom tujuan untuk menulis data. Tentukan nama kolom dalam bentuk array, misalnya, `"column":["id","name","age"]`. Untuk menulis data ke semua kolom secara berurutan, gunakan tanda bintang (``) dalam array, misalnya, `"column":[""]`.	Ya	Tidak ada nilai default
preSql	Pernyataan SQL yang akan dijalankan sebelum tugas sinkronisasi data dimulai. Di Antarmuka tanpa kode, Anda hanya dapat menjalankan satu pernyataan SQL. Di editor kode, Anda dapat menjalankan beberapa pernyataan SQL. Misalnya, Anda dapat menggunakan pernyataan ini untuk menghapus data yang sudah ada dari tabel.	Tidak	Tidak ada nilai default
postSql	Pernyataan SQL yang ingin dijalankan setelah tugas sinkronisasi dijalankan. Misalnya, Anda dapat mengatur parameter ini ke pernyataan SQL yang digunakan untuk menambahkan timestamp. Anda hanya dapat menjalankan satu pernyataan SQL di Antarmuka tanpa kode dan beberapa pernyataan SQL di editor kode.	Tidak	Tidak ada nilai default
maxBatchRows	Jumlah maksimum baris yang dapat Anda tulis ke tabel tujuan sekaligus. Parameter ini dan parameter batchSize menentukan jumlah catatan data yang dapat Anda tulis ke tabel tujuan sekaligus. Setiap kali data yang di-cache mencapai nilai salah satu parameter tersebut, writer mulai menulis data ke tabel tujuan.	Tidak	500000
batchSize	Jumlah maksimum data yang dapat Anda tulis ke tabel tujuan sekaligus. Parameter ini dan parameter maxBatchRows menentukan jumlah catatan data yang dapat Anda tulis ke tabel tujuan sekaligus. Setiap kali data yang di-cache mencapai nilai salah satu parameter tersebut, writer mulai menulis data ke tabel tujuan.	Tidak	104857600
maxRetries	Jumlah maksimum percobaan ulang yang diizinkan setelah Anda gagal menulis beberapa catatan data ke tabel tujuan sekaligus.	Tidak	3
labelPrefix	Awalan label untuk setiap batch file yang diunggah. Label akhir merupakan kombinasi dari `labelPrefix + UUID` untuk memastikan setiap impor bersifat idempoten, mencegah duplikasi data.	Tidak	datax_doris_writer_
loadProps	Parameter permintaan untuk StreamLoad, terutama digunakan untuk mengonfigurasi format data impor. Secara default, data diimpor dalam format CSV. Jika parameter loadProps tidak dikonfigurasi, format CSV default digunakan, dengan `\t` sebagai pemisah kolom dan `\n` sebagai pemisah baris. Konfigurasi default adalah sebagai berikut: `"loadProps": { "format":"csv", "column_separator": "\t", "line_delimiter": "\n" }` Jika Anda ingin menulis data dalam format JSON, gunakan pengaturan berikut: `"loadProps": { "format": "json" }`	Tidak	Tidak ada nilai default

Menulis data tipe agregat

Anda dapat menggunakan Doris Writer untuk menulis data ke kolom dengan tipe agregasi tertentu. Dalam hal ini, Anda harus mengonfigurasi parameter tambahan.

Misalnya, pada tabel Doris berikut, uuid bertipe bitmap (tipe agregasi) dan sex bertipe HLL (tipe agregasi).

CREATE TABLE `example_table_1` (
  `user_id` int(11) NULL,
  `date` varchar(10) NULL DEFAULT "10.5",
  `city` varchar(10) NULL,
  `uuid` bitmap BITMAP_UNION NULL, -- Tipe agregasi
  `sex` HLL HLL_UNION  -- Tipe agregasi
) ENGINE=OLAP AGGREGATE KEY(`user_id`, `date`,`city`)
COMMENT 'OLAP' DISTRIBUTED BY HASH(`user_id`) BUCKETS 32

Masukkan data mentah ke dalam tabel:

user_id,date,city,uuid,sex
0,T0S4Pb,abc,43,'54'
1,T0S4Pd,fsd,34,'54'
2,T0S4Pb,fa3,53,'64'
4,T0S4Pb,fwe,87,'64'
5,T0S4Pb,gbr,90,'56'
2,iY3GiHkLF,234,100,'54'

Saat menggunakan Doris Writer untuk menulis data ke kolom bertipe agregasi, Anda harus menentukan kolom tersebut di writer.parameter.column dan mengonfigurasi fungsi agregat di writer.parameter.loadProps.columns. Misalnya, gunakan fungsi agregat bitmap_hash untuk kolom uuid dan fungsi agregat hll_hash untuk kolom sex.

Kode contoh:

{
    "stepType": "doris",// Nama plug-in. 
    "writer":
    {
        "parameter":
        {
            "column":
            [
                "user_id",
                "date",
                "city",
                "uuid",// Tipe agregasi adalah bitmap.
                "sex"// Tipe agregasi adalah HLL.
            ],
            "loadProps":
            {
                "format": "csv",
                "column_separator": "\\x01",
                "line_delimiter": "\\x02",
                "columns": "user_id,date,city,k1,uuid=bitmap_hash(k1),k2,sex=hll_hash(k2)"// Anda harus menentukan fungsi agregat.
            },
            "postSql":
            [
                "select count(1) from example_tbl_3"
            ],
            "preSql":
            [],
            "datasource":"doris_datasource",// Nama sumber data. 
                    "table": "doris_table_name",// Nama tabel. 
        }
          "name": "Writer",
              "category": "writer"
    }
}