Gunakan Data Integration untuk menyinkronkan data dari MongoDB ke MaxCompute secara batch - DataWorks

DataWorks Data Integration menyediakan plugin MongoDB Reader untuk membaca data dari MongoDB dan menyinkronkannya ke sumber data lain. Topik ini menjelaskan cara menggunakan Data Integration untuk menyinkronkan data dari MongoDB ke MaxCompute secara batch.

Latar Belakang

Dalam tutorial ini, sumber datanya adalah MongoDB dan tujuannya adalah MaxCompute. Sebelum memulai, siapkan data MongoDB dan buat tabel tujuan di MaxCompute.

Prasyarat

Sebelum memulai, pastikan Anda memenuhi persyaratan berikut:

Anda telah mengaktifkan DataWorks dan membuat sumber data MaxCompute.
Tutorial ini menggunakan grup sumber daya eksklusif untuk Data Integration guna menjalankan task batch. Anda harus membeli dan mengonfigurasi grup sumber daya eksklusif tersebut. Untuk informasi selengkapnya, lihat Use an exclusive resource group for Data Integration.

Catatan
Anda juga dapat menggunakan grup sumber daya serverless. Untuk informasi selengkapnya, lihat Use a serverless resource group.

Siapkan data sampel dan tabel

Tutorial ini memerlukan koleksi MongoDB dan tabel MaxCompute untuk sinkronisasi batch.

Siapkan koleksi MongoDB.

Tutorial ini menggunakan ApsaraDB for MongoDB sebagai contoh. Kode berikut menunjukkan cara menyiapkan koleksi MongoDB.

Buat koleksi bernama di_mongodb_conf_test.

db.createCollection('di_mongodb_conf_test')

Masukkan data sampel untuk tutorial ini ke dalam koleksi.

db.di_mongodb_conf_test.insertOne({
    'col_string':'mock string value',
    'col_int32':NumberInt("1"),
    'col_int32_min':NumberInt("-2147483648"),
    'col_int32_max':NumberInt("2147483647"),
    'col_int64':NumberLong("1234567890123456"),
    'col_int64_min':NumberLong("-9223372036854775807"),
    'col_int64_max':NumberLong("9223372036854775807"),
    'col_decimal':NumberDecimal("9999999.4999999999"),
    'col_double':9999999.99,
    'col_boolean':true,
    'col_timestamp':ISODate(),
    'col_date':new Date(),
    'col_array_to_json':['a','b'],
    'col_array_to_join':['a','b'],
    'col_doc':{
        'key_string':'mock string value',
        'key_int32':NumberInt("1"),
        'key_int32_min':NumberInt("-2147483648"),
        'key_int32_max':NumberInt("2147483647"),
        'key_int64':NumberLong("1234567890123456"),
        'key_int64_min':NumberLong("-9223372036854775807"),
        'key_int64_max':NumberLong("9223372036854775807"),
        'key_decimal':NumberDecimal("9999999.4999999999"),
        'key_double':9999999.99,
        'key_boolean':true,
        'key_timestamp':ISODate(),
        'key_date':new Date(),
        'key_array_to_json':['a','b'],
        'key_array_to_join':['a','b'],
    },
    'col_extra_1':'this is extra 1',
    'col_extra_2':'this is extra 2',
})

Kueri data yang dimasukkan ke MongoDB.
```
db.getCollection("di_mongodb_conf_test").find({})
```
Hasil kueri menunjukkan satu dokumen uji di dalam koleksi, dengan _id bernilai 63dca714b8548a78e1dc3238. Dokumen ini berisi berbagai tipe data: string (col_string), integer (col_int32/col_int64), floating-point (col_double/col_decimal), Boolean (col_boolean), tanggal/waktu (col_date/col_timestamp), dan array (col_array_to_join/col_array_to_json). Dokumen ini juga mencakup dokumen bersarang, col_doc, serta dua bidang tambahan, col_extra_1 dan col_extra_2.

Siapkan tabel MaxCompute.

Buat tabel partisi bernama di_mongodb_conf_test dengan pt sebagai bidang partisi.

CREATE TABLE IF NOT EXISTS di_mongodb_conf_test
(
  `id`                 STRING
  ,`col_string`        STRING
  ,`col_int32`         INT
  ,`col_int32_min`     INT
  ,`col_int32_max`     INT
  ,`col_int64`         BIGINT
  ,`col_int64_min`     BIGINT
  ,`col_int64_max`     BIGINT
  ,`col_decimal`       DECIMAL(38,18)
  ,`col_double`        DOUBLE
  ,`col_boolean`       BOOLEAN
  ,`col_timestamp`     TIMESTAMP
  ,`col_date`          DATE
  ,`col_array_to_json` STRING
  ,`col_array_to_join` STRING
  ,`key_string`        STRING
  ,`key_int32`         INT
  ,`key_int32_min`     INT
  ,`key_int32_max`     INT
  ,`key_int64`         BIGINT
  ,`key_int64_min`     BIGINT
  ,`key_int64_max`     BIGINT
  ,`key_decimal`       DECIMAL(38,18)
  ,`key_double`        DOUBLE
  ,`key_boolean`       BOOLEAN
  ,`key_timestamp`     TIMESTAMP
  ,`key_date`          DATE
  ,`key_array_to_json` STRING
  ,`key_array_to_join` STRING
  ,`col_doc`           STRING
  ,`col_combine`       STRING
)
PARTITIONED BY
(
  pt                   STRING
)
LIFECYCLE 36500
;

Tambahkan partisi dengan nilai 20230202.

alter table di_mongodb_conf_test add if not exists partition (pt='20230202');

Verifikasi bahwa tabel partisi telah dibuat dengan benar.
```
SELECT*FROM di_mongodb_conf_test
WHEREpt='20230202';
```

Konfigurasikan task sinkronisasi batch

Langkah 1: Tambahkan sumber data MongoDB

Tambahkan sumber data MongoDB dan pastikan konektivitas jaringan antara sumber data dan grup sumber daya eksklusif untuk Data Integration. Untuk informasi selengkapnya, lihat Add a MongoDB data source.

Langkah 2: Buat dan konfigurasikan node sinkronisasi batch

Di DataWorks DataStudio, buat node sinkronisasi batch dan konfigurasikan sumber serta tujuannya. Bagian ini menyoroti parameter utama; Anda dapat menggunakan nilai default untuk parameter lainnya. Untuk instruksi terperinci, lihat Configure a batch synchronization node using the Codeless UI.

Konfigurasikan koneksi jaringan.

Pilih sumber data MongoDB dan MaxCompute, grup sumber daya eksklusif untuk Data Integration yang sesuai, lalu uji konektivitasnya.
Konfigurasikan task: Pilih sumber data.

Untuk sumber dan tujuan, pilih koleksi MongoDB dan tabel MaxCompute yang dipartisi.

Konfigurasikan task: Pemetaan bidang.

Saat sumber datanya adalah MongoDB, Peer mapping digunakan secara default. Anda juga dapat mengklik ikon icon untuk mengedit bidang tabel sumber secara manual. Berikut ini contoh pengeditan manual.

{"name":"_id","type":"string"}
{"name":"col_string","type":"string"}
{"name":"col_int32","type":"long"}
{"name":"col_int32_min","type":"long"}
{"name":"col_int32_max","type":"long"}
{"name":"col_int64","type":"long"}
{"name":"col_int64_min","type":"long"}
{"name":"col_int64_max","type":"long"}
{"name":"col_decimal","type":"double"}
{"name":"col_double","type":"double"}
{"name":"col_boolean","type":"boolean"}
{"name":"col_timestamp","type":"date"}
{"name":"col_date","type":"date"}
{"name":"col_array_to_json","type":"string"}
{"name":"col_array_to_join","type":"array","splitter":","}
{"name":"col_doc.key_string","type":"document.string"}
{"name":"col_doc.key_int32","type":"document.long"}
{"name":"col_doc.key_int32_min","type":"document.long"}
{"name":"col_doc.key_int32_max","type":"document.long"}
{"name":"col_doc.key_int64","type":"document.long"}
{"name":"col_doc.key_int64_min","type":"document.long"}
{"name":"col_doc.key_int64_max","type":"document.long"}
{"name":"col_doc.key_decimal","type":"document.double"}
{"name":"col_doc.key_double","type":"document.double"}
{"name":"col_doc.key_boolean","type":"document.boolean"}
{"name":"col_doc.key_timestamp","type":"document.date"}
{"name":"col_doc.key_date","type":"document.date"}
{"name":"col_doc.key_array_to_json","type":"document"}
{"name":"col_doc.key_array_to_join","type":"document.array","splitter":","}
{"name":"col_doc","type":"string"}
{"name":"col_combine","type":"combine"}

Setelah Anda mengedit bidang-bidang tersebut, antarmuka pengguna akan menampilkan pemetaan antara bidang sumber dan tujuan.

Langkah 3: Commit dan deploy node

Jika Anda menggunakan ruang kerja mode standar dan ingin menjadwalkan task ini, commit dan deploy node ke lingkungan produksi. Untuk informasi selengkapnya, lihat Deploy tasks.

Langkah 4: Jalankan node dan lihat hasilnya

Setelah mengonfigurasi node, jalankan node tersebut. Saat task selesai, lihat data yang telah disinkronkan di tabel MaxCompute. Data yang disinkronkan mencakup bidang dan nilai berikut: col_string (mock string value), col_int32 (1), col_int32_min (-2147483648), col_int32_max (2147483647), col_int64 (1234567890123456), col_decimal (9999999.4999999999), col_double (9999999.99), col_boolean (true), col_timestamp (2023-02-03 14:17:56.554), col_date (2023-02-03), col_array_to_json ([a, b]), col_array_to_join (a,b), dan pt (20230202). Beberapa bidang numerik disimpan sebagai teks. Nilai untuk bidang kompleks adalah sebagai berikut:

Konten bidang col_doc adalah sebagai berikut.

{
  "key_array_to_join": [
    "a",
    "b"
  ],
  "key_array_to_json": [
    "a",
    "b"
  ],
  "key_boolean": true,
  "key_date": "2023-02-03 14:17:56",
  "key_decimal": {
    "finite": true,
    "high": 34711494127958097792,
    "infinite": false,
    "low": 99999994999999999,
    "naN": false,
    "negative": false
  },
  "key_double": 9999999.99,
  "key_int32": 1,
  "key_int32_max": 2147483647,
  "key_int32_min": -2147483648,
  "key_int64": 1234567890123456,
  "key_int64_max": 9223372036854775807,
  "key_int64_min": -9223372036854775807,
  "key_string": "mock string value",
  "key_timestamp": "2023-02-03 14:17:56"
}

Konten bidang col_combine adalah sebagai berikut.

{
  "col_extra_1": "this is extra 1",
  "col_extra_2": "this is extra 2"
}

Catatan

Untuk informasi tentang masalah terkait output data tipe Decimal, lihat Appendix 2: Handle the output of the Decimal type in documents.

Lampiran 1: Konversi format data

Konversi data array ke format JSON: `col_array_to_json`

Source MongoDB data

Field mapping configuration

Output to MaxCompute

{
    "col_array_to_json":
    [
        "a",
        "b"
    ]
}

{"name":"col_array_to_json","type":"string"}

Dalam konfigurasi pemetaan bidang, jika type bernilai string, task sinkronisasi akan melakukan serialisasi data asli ke dalam format JSON untuk output saat waktu proses.

[a, b]

Konversi array menjadi string gabungan: `col_array_to_join`

Source MongoDB data

Field mapping configuration

Output to MaxCompute

{
    "col_array_to_join":
    [
        "a",
        "b"
    ]
}

{"name":"col_array_to_join","type":"array","splitter":","}

Saat mengonfigurasi pemetaan bidang, parameter splitter wajib diisi jika type bernilai array. Saat task sinkronisasi dijalankan, konten array data sumber akan digabung menggunakan splitter, dan output akhirnya adalah string hasil penggabungan tersebut.

a,b

Sinkronkan bidang dokumen bersarang

Source MongoDB data

Field mapping configuration

Output to MaxCompute

{
    "col_doc":
    {
        "key_string": "mock string value"
    }
}

{"name":"col_doc.key_string","type":"document.string"}

name menentukan path bidang yang akan disinkronkan dalam document. Saat task sinkronisasi dijalankan, sistem akan membaca document berdasarkan path tersebut dan mengeluarkan datanya.

mock string value

Serialisasi dokumen sebagai string JSON

Source MongoDB data

Field mapping configuration

Output to MaxCompute

{
    "col_doc":
    {
        "key_string": "mock string value",
        "key_int32": 1
    }
}

{"name":"col_doc","type":"string"}

Saat mengonfigurasi pemetaan bidang, jika type bernilai string, task sinkronisasi akan melakukan serialisasi seluruh col_doc menjadi string JSON dan mengeluarkannya saat waktu proses.

{"key_string":"mockstringvalue","key_int32":1}

Serialisasi bidang yang tidak dipetakan sebagai JSON

Source MongoDB data

Field mapping configuration

Output to MaxCompute

{
    "col_1": "value1",
    "col_2": "value2",
    "col_3": "value3",
    "col_4": "value4"
}

{"name":"col_1","type":"string"}
{"name":"col_2","type":"string"}
{"name":"col_combine","type":"combine"}

Dokumen ini memiliki empat bidang. col_1 dan col_2 dipetakan secara eksplisit. Selama sinkronisasi, bidang yang tidak dipetakan (col_3 dan col_4) akan diserialisasi menjadi objek JSON dan dikeluarkan.

{"col_3":"value3","col_4":"value4"}

Lampiran 2: Output tipe Decimal

Saat dokumen diserialisasi ke format JSON, data Decimal128 secara default dikeluarkan sebagai berikut:

{
    "key_decimal":
    {
        "finite": true,
        "high": 3471149412795809792,
        "infinite": false,
        "low": 99999994999999999,
        "naN": false,
        "negative": false
    }
}

Untuk mengeluarkan data sebagai tipe numerik, ikuti langkah-langkah berikut:

Saat mengonfigurasi task sinkronisasi batch, beralihlah ke mode skrip.

Ubah konfigurasi task Reader: pada bagian parameter, tambahkan parameter decimal128OutputType dan atur nilainya menjadi bigDecimal.

"parameter": {
    "collectionName": "di_mongodb_conf_test",
    "decimal128OutputType":"bigDecimal"
},
"name": "Reader",
"category": "reader"

Jalankan kembali task sinkronisasi batch dan lihat hasilnya.
```
{
    "key_decimal": "9999999.4999999999"
}
```

Latar Belakang

Prasyarat

Siapkan data sampel dan tabel

Konfigurasikan task sinkronisasi batch

Langkah 1: Tambahkan sumber data MongoDB

Langkah 2: Buat dan konfigurasikan node sinkronisasi batch

Langkah 3: Commit dan deploy node

Langkah 4: Jalankan node dan lihat hasilnya

Lampiran 1: Konversi format data

Konversi data array ke format JSON: col_array_to_json

Konversi array menjadi string gabungan: col_array_to_join

Sinkronkan bidang dokumen bersarang

Serialisasi dokumen sebagai string JSON

Serialisasi bidang yang tidak dipetakan sebagai JSON

Lampiran 2: Output tipe Decimal

Konversi data array ke format JSON: `col_array_to_json`

Konversi array menjadi string gabungan: `col_array_to_join`