DataWorks：Gunakan tugas sinkronisasi batch untuk membaca data dari sumber data MongoDB

Prasyarat

Persiapan awal

Dalam contoh ini, Anda perlu menyiapkan koleksi data MongoDB dan tabel MaxCompute untuk sinkronisasi data.

1. Siapkan Koleksi Data MongoDB.

   Dalam contoh ini, ApsaraDB for MongoDB digunakan. Contoh kode berikut menunjukkan cara menyiapkan koleksi data ApsaraDB for MongoDB.

   1. Buat koleksi data bernama di_mongodb_conf_test.

      db.createCollection('di_mongodb_conf_test')

   2. Masukkan data sampel ke dalam koleksi data.

      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',
      })

   3. Kueri data yang dimasukkan ke dalam koleksi data MongoDB.

      db.getCollection("di_mongodb_conf_test").find({})

      Gambar berikut menunjukkan hasil kueri.

2. Siapkan Tabel MaxCompute.

   1. Buat tabel partisi bernama di_mongodb_conf_test. Bidang partisi adalah pt.

      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
      ;

   2. Tambahkan nilai 20230202 ke bidang partisi.

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

   3. Periksa apakah tabel partisi dibuat dengan benar.

      SELECT*FROM di_mongodb_conf_test
      WHEREpt='20230202';

Konfigurasikan tugas sinkronisasi batch

Langkah 1: Tambahkan sumber data MongoDB

Tambahkan sumber data MongoDB dan pastikan koneksi jaringan telah dibuat antara sumber data dan grup sumber daya eksklusif untuk Data Integration. Untuk informasi lebih lanjut, lihat Tambahkan Sumber Data MongoDB.

Langkah 2: Buat dan konfigurasikan tugas sinkronisasi batch

Langkah 3: Kirim dan terapkan tugas sinkronisasi batch

Jika Anda menggunakan ruang kerja dalam mode standar dan ingin menjadwalkan tugas sinkronisasi batch secara berkala di lingkungan produksi, Anda dapat mengirim dan menerapkan tugas ke lingkungan produksi. Untuk informasi lebih lanjut, lihat Terapkan Node.

Langkah 4: Jalankan tugas sinkronisasi batch dan lihat hasil sinkronisasi

Lampiran 1: Konversi tipe data selama sinkronisasi data

Konversi array menjadi data JSON: col_array_to_json

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

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

[a, b]

Konversi array menjadi string gabungan: col_array_to_join

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

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

a,b

Baca data dari bidang bersarang tertentu dalam dokumen MongoDB

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

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

mock string value

Konversi data dalam dokumen MongoDB menjadi data JSON

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

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

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

Konversi data dalam bidang kecuali bidang yang dikonfigurasikan dalam dokumen MongoDB menjadi data JSON

{
    "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"}

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

Lampiran 2: Masalah terkait data keluaran yang dikonversi dari data tipe DECIMAL

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