DataWorks:Baca data dari sumber data MongoDB melalui sinkronisasi batch

Prasyarat

Siapkan tabel data sampel

Siapkan koleksi data MongoDB dan tabel MaxCompute untuk sinkronisasi data offline.

1. Siapkan koleksi data MongoDB.

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

   1. Buat koleksi data bernama di_mongodb_conf_test.

      db.createCollection('di_mongodb_conf_test')

   2. Masukkan data sampel untuk tutorial ini 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 MongoDB.

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

      Hasil kueri ditampilkan pada gambar berikut:

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 partisi dengan nilai 20230202.

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

   3. Verifikasi bahwa tabel partisi telah dibuat.

      SELECT*FROM di_mongodb_conf_test
      WHEREpt='20230202';

Konfigurasikan task offline

Langkah 1: Tambahkan sumber data MongoDB

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

Langkah 2: Buat node sync offline dan konfigurasikan task sinkronisasi

Langkah 3: Kirim dan publikasikan node sync offline

Jika Anda menggunakan ruang kerja DataWorks mode standar dan ingin menjadwalkan task sync offline agar berjalan secara berkala di lingkungan produksi, Anda harus mengirim dan mempublikasikan node tersebut. Untuk informasi selengkapnya, lihat Publish a task.

Langkah 4: Jalankan node sync offline dan lihat hasilnya

Lampiran 1: Transformasi format data selama sinkronisasi

Konversi data array ke format JSON: col_array_to_json

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

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

[a, b]

Konversi data 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 bidang tertentu dari dokumen bersarang

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

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

mock string value

Data dokumen sebagai output serialisasi JSON

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

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

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

Serialisasi sisa bidang dokumen ke format 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: Output tipe Decimal dalam dokumen

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