DataWorks：Sumber data MongoDB

Tipe data MongoDB yang didukung oleh MongoDB Reader

Data Integration mendukung sebagian besar tipe data MongoDB. Namun, beberapa tipe data tidak didukung. Verifikasi tipe data Anda sebelum melanjutkan.

Untuk tipe data yang didukung, Data Integration membaca data sebagai berikut:

• Untuk tipe data primitif, Data Integration secara otomatis membaca data dari path yang sesuai berdasarkan nama bidang baca (column) yang dikonfigurasi untuk tugas sinkronisasi. Data Integration juga secara otomatis mengonversi tipe data tersebut. Anda tidak perlu menentukan properti type untuk kolom tersebut. Untuk informasi lebih lanjut, lihat Lampiran: Contoh skrip MongoDB dan deskripsi parameter.

  Type	Baca offline (MongoDB Reader)	Deskripsi
  ObjectId	Didukung	Tipe object ID.
  Double	Dukungan	Tipe bilangan titik mengambang 64-bit.
  Integer 32-bit	Dukungan	Integer 32-bit.
  Integer 64-bit	Dukungan	Integer 64-bit.
  Decimal128	Dukungan	Tipe Decimal128. Catatan Jika tipe ini dikonfigurasi sebagai tipe nested atau tipe Combine, data diproses sebagai objek selama serialisasi JSON. Anda harus menambahkan parameter decimal128OutputType dan mengaturnya ke bigDecimal untuk mengeluarkan data sebagai desimal.
  String	Didukung	Tipe string.
  Boolean	Dukungan	Tipe Boolean.
  Timestamp	Dukungan	Tipe timestamp. Catatan BsonTimestamp menyimpan timestamp. Anda tidak perlu mempertimbangkan dampak zona waktu. Untuk informasi lebih lanjut, lihat Masalah Zona Waktu di MongoDB.
  Date	Dukungan	Tipe tanggal.

• Untuk beberapa tipe data kompleks, Anda dapat menyesuaikan pemrosesan dengan mengonfigurasi properti type dari kolom tersebut.

  Type	Baca offline (MongoDB Reader)	Deskripsi
  Document	Dukungan	Tipe dokumen tersemat. Jika properti type tidak dikonfigurasi, dokumen langsung diproses melalui serialisasi JSON. Jika properti type diatur ke document, ini merupakan tipe nested. MongoDB Reader membaca properti dokumen berdasarkan path. Untuk contoh detail, lihat Contoh tipe data 2: Mengurai dokumen nested multi-level secara rekursif.
  Array	Dukungan	Tipe array. Jika type diatur ke array.json atau arrays, data langsung diproses melalui serialisasi JSON. Jika type diatur ke array atau document.array, data digabung menjadi string. Pemisah default (splitter dalam kolom) adalah koma (,).

  Penting

  Data Integration tidak mendukung tipe array. Namun, MongoDB mendukung array dan fitur pengindeksannya yang canggih. Anda dapat mengonfigurasi parameter tertentu untuk mengonversi string menjadi array di MongoDB. Setelah konversi, Anda dapat menulis data ke MongoDB secara paralel.

Tipe data khusus Data Integration: combine

Konversi tipe data MongoDB Reader

Tabel berikut mencantumkan konversi tipe data yang dilakukan oleh MongoDB Reader.

Konversi tipe data MongoDB Writer

Contoh tipe data 1: Menggunakan tipe combine

Tipe data Combine dari plugin MongoDB Reader memungkinkan Anda menggabungkan beberapa bidang dalam dokumen MongoDB menjadi satu string JSON. Misalnya, Anda ingin mengimpor bidang dari MongoDB ke MaxCompute. Tiga dokumen berikut berisi bidang di mana kunci merepresentasikan seluruh bidang dan nilainya dihilangkan. Bidang `a` dan `b` umum di semua dokumen, sedangkan `x_n` adalah bidang non-tetap.

• doc1: a b x_1 x_2

• doc2: a b x_2 x_3 x_4

• doc3: a b x_5

Dalam file konfigurasi, tentukan bidang yang memerlukan pemetaan satu-ke-satu. Untuk bidang yang akan digabung, berikan nama baru yang berbeda dari bidang apa pun yang ada dalam dokumen, dan atur tipe ke COMBINE, seperti yang ditunjukkan di bawah ini.

"column": [
{
"name": "a",
"type": "string",
},
{
"name": "b",
"type": "string",
},
{
"name": "doc",
"type": "combine",
}
]

Hasil akhir yang diekspor ke MaxCompute adalah sebagai berikut.

Contoh tipe data 2: Mengurai dokumen nested multi-level secara rekursif

Ketika dokumen MongoDB memiliki beberapa level nesting, Anda dapat mengonfigurasi tipe document untuk memprosesnya secara rekursif. Berikut adalah contohnya:

• Data sumber di MongoDB:

  {
      "name": "name1",
      "a":
      {
          "b":
          {
              "c": "this is value"
          }
      }
  }

• Konfigurasi kolom MongoDB:

  {"name":"_id","type":"string"}
  {"name":"name","type":"string"}
  {"name":"a.b.c","type":"document"}

  

Dengan konfigurasi ini, nilai bidang nested `a.b.c` dari sumber ditulis ke bidang tujuan `c`. Setelah tugas sinkronisasi dijalankan, data yang ditulis ke tujuan adalah this is value.

Konfigurasi tugas sinkronisasi offline tabel tunggal

• Untuk prosedurnya, lihat Konfigurasi tugas di Antarmuka tanpa kode dan Konfigurasi tugas di editor kode.

• Untuk semua parameter dan contoh skrip konfigurasi tugas di editor kode, lihat Lampiran: Contoh skrip MongoDB dan deskripsi parameter.

Konfigurasi tugas sinkronisasi real-time tabel tunggal

Untuk prosedurnya, lihat Konfigurasi tugas sinkronisasi real-time di Data Integration dan Konfigurasi tugas sinkronisasi real-time di DataStudio.

Konfigurasi tugas sinkronisasi seluruh database

Untuk informasi tentang cara mengonfigurasi tugas sinkronisasi seluruh database, seperti sinkronisasi offline seluruh database, sinkronisasi real-time penuh dan inkremental seluruh database, dan sinkronisasi real-time seluruh database dengan sharding, lihat Tugas sinkronisasi offline seluruh database dan Konfigurasi tugas sinkronisasi real-time seluruh database.

Konfigurasi tugas sinkronisasi batch menggunakan editor kode

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

Contoh skrip Reader

Kode berikut menunjukkan cara mengonfigurasi pekerjaan untuk mengekstrak data dari MongoDB ke tujuan lokal. Untuk informasi lebih lanjut tentang parameter, lihat tabel berikut.

{
    "type":"job",
    "version":"2.0",// Nomor versi.
    "steps":[
        {
            "category": "reader",
            "name": "Reader",
            "parameter": {
                "datasource": "datasourceName", // Nama sumber data.
                "collectionName": "tag_data", // Nama koleksi.
                "query": "", // Filter kueri data.
                "column": [
                    {
                        "name": "unique_id", // Nama bidang.
                        "type": "string" // Tipe bidang.
                    },
                    {
                        "name": "sid",
                        "type": "string"
                    },
                    {
                        "name": "user_id",
                        "type": "string"
                    },
                    {
                        "name": "auction_id",
                        "type": "string"
                    },
                    {
                        "name": "content_type",
                        "type": "string"
                    },
                    {
                        "name": "pool_type",
                        "type": "string"
                    },
                    {
                        "name": "frontcat_id",
                        "type": "array",
                        "splitter": ""
                    },
                    {
                        "name": "categoryid",
                        "type": "array",
                        "splitter": ""
                    },
                    {
                        "name": "gmt_create",
                        "type": "string"
                    },
                    {
                        "name": "taglist",
                        "type": "array",
                        "splitter": " "
                    },
                    {
                        "name": "property",
                        "type": "string"
                    },
                    {
                        "name": "scorea",
                        "type": "int"
                    },
                    {
                        "name": "scoreb",
                        "type": "int"
                    },
                    {
                        "name": "scorec",
                        "type": "int"
                    },
                    {
                        "name": "a.b",
                        "type": "document.int"
                    },
                    {
                        "name": "a.b.c",
                        "type": "document.array",
                        "splitter": " "
                    }
                ]
            },
            "stepType": "mongodb"
        },
        { 
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "common": { 
            "column": { 
                "timeZone": "GMT+0" // Zona waktu.
            } 
        },
        "errorLimit":{
            "record":"0"// Jumlah catatan error.
        },
        "speed":{
            "throttle":true,// Jika throttle diatur ke false, parameter mbps tidak berlaku, yang berarti laju data tidak dibatasi. Jika throttle diatur ke true, laju data dibatasi.
            "concurrent":1, // Jumlah pekerjaan konkuren.
            "mbps":"12"// Batas laju data. 1 mbps setara dengan 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Contoh skrip Writer

Kode berikut menunjukkan cara mengonfigurasi pekerjaan sinkronisasi data untuk menulis data ke MongoDB. Untuk informasi lebih lanjut tentang parameter, lihat tabel berikut.

{
    "type": "job",
    "version": "2.0",// Nomor versi.
    "steps": [
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "mongodb",// Nama plug-in.
            "parameter": {
                "datasource": "",// Nama sumber data.
                "column": [
                    {
                        "name": "_id",// Nama kolom.
                        "type": "ObjectId"// Tipe data. Jika replaceKey diatur ke _id, tipe harus diatur ke ObjectId. Jika Anda mengatur tipe ke string, penggantian gagal.
                    },
                    {
                        "name": "age",
                        "type": "int"
                    },
                    {
                        "name": "id",
                        "type": "long"
                    },
                    {
                        "name": "wealth",
                        "type": "double"
                    },
                    {
                        "name": "hobby",
                        "type": "array",
                        "splitter": " "
                    },
                    {
                        "name": "valid",
                        "type": "boolean"
                    },
                    {
                        "name": "date_of_join",
                        "format": "yyyy-MM-dd HH:mm:ss",
                        "type": "date"
                    }
                ],
                "writeMode": {// Mode penulisan.
                    "isReplace": "true",
                    "replaceKey": "_id"
                },
                "collectionName": "datax_test"// Nama koneksi.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {// Jumlah catatan error.
            "record": "0"
        },
        "speed": {
            "throttle": true,// Jika throttle diatur ke false, parameter mbps tidak berlaku, yang berarti laju data tidak dibatasi. Jika throttle diatur ke true, laju data dibatasi.
            "concurrent": 1,// Jumlah pekerjaan konkuren.
            "mbps": "1"// Batas laju data. 1 mbps setara dengan 1 MB/s.
        },
       "jvmOption": "-Xms1024m -Xmx1024m"
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Parameter skrip Writer

After applying the update, the (immutable) field '_id' was found to have been altered to _id: "2"

Saat pekerjaan Data Integration dijalankan, preSql yang dikonfigurasi dieksekusi terlebih dahulu. Fase penulisan data aktual baru dimulai setelah preSql selesai. Parameter preSql sendiri tidak memengaruhi konten data yang ditulis. Parameter preSql memungkinkan Data Integration mendukung eksekusi idempoten. Misalnya, preSql Anda dapat membersihkan data historis sebelum setiap eksekusi tugas sesuai aturan bisnis Anda. Dalam hal ini, jika tugas gagal, Anda hanya perlu menjalankan ulang pekerjaan Data Integration.

Persyaratan format untuk preSql adalah sebagai berikut:

• Anda harus mengonfigurasi field type untuk menunjukkan kategori operasi awal. Nilai yang didukung adalah drop dan remove. Misalnya, "preSql":{"type":"remove"}.

  • drop: Menghapus koleksi dan datanya. Koleksi yang ditentukan oleh parameter collectionName adalah koleksi yang akan dihapus.

  • remove: Menghapus data berdasarkan kondisi.

  • json: Anda dapat menggunakan JSON untuk menentukan kondisi penghapusan data. Misalnya, "preSql":{"type":"remove", "json":"{'operationTime':{'$gte':ISODate('${last_day}T00:00:00.424+0800')}}"}. Di sini, ${last_day} adalah parameter penjadwalan DataWorks dalam format $[yyyy-mm-dd]. Anda dapat menggunakan operator kondisional MongoDB yang didukung (seperti $gt, $lt, $gte, dan $lte), operator logika (seperti and dan or), atau fungsi (seperti max, min, sum, avg, dan ISODate) sesuai kebutuhan.

    Data Integration mengeksekusi operasi data Anda untuk menghapus kueri menggunakan API MongoDB standar berikut.

    query=(BasicDBObject) com.mongodb.util.JSON.parse(json);        
    col.deleteMany(query);

    Catatan

    Untuk menghapus data secara kondisional, kami menyarankan menggunakan format konfigurasi JSON.

  • item: Anda dapat mengonfigurasi nama kolom (name), kondisi (condition), dan nilai kolom (value) untuk penyaringan data dalam item. Misalnya, "preSql":{"type":"remove","item":[{"name":"pv","value":"100","condition":"$gt"},{"name":"pid","value":"10"}]}.

    Data Integration membuat kondisi kueri berdasarkan kondisi item yang Anda konfigurasi, lalu mengeksekusi penghapusan menggunakan API MongoDB standar. Misalnya, col.deleteMany(query);.

• Jika preSql tidak dikenali, tidak ada operasi penghapusan awal yang dilakukan.