DataWorks:Sumber data MongoDB

Tipe data MongoDB yang didukung oleh MongoDB Reader

Data Integration mendukung sebagian besar, tetapi tidak semua, tipe data MongoDB. Pastikan tipe data Anda didukung.

Ketika Data Integration membaca tipe data yang didukung, operasi berikut dilakukan:

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

  Type	Baca batch (MongoDB Reader)	Deskripsi
  ObjectId	Didukung	Tipe object ID.
  Double	Didukung	Tipe bilangan titik mengambang 64-bit.
  Integer 32-bit	Didukung	Integer 32-bit.
  Integer 64-bit	Didukung	Integer 64-bit.
  Decimal128	Didukung	Tipe Decimal128. Catatan Jika suatu bidang dikonfigurasi sebagai tipe nested atau tipe combine, bidang tersebut 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	Didukung	Tipe Boolean.
  Timestamp	Didukung	Tipe timestamp. Catatan BsonTimestamp menyimpan stempel waktu. 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 mengonfigurasi properti type untuk kolom tersebut guna melakukan pemrosesan kustom.

  Type	Baca batch (MongoDB Reader)	Deskripsi
  Document	Didukung	Tipe dokumen tersemat. Jika properti type tidak dikonfigurasi, Document langsung dikonversi menggunakan serialisasi JSON. Jika properti type diatur ke document, bidang tersebut merupakan tipe nested. MongoDB Reader membaca properti Document berdasarkan path. Untuk contoh detail, lihat Contoh 2: Mengurai secara rekursif Document bersarang multi-level di bawah.
  Array	Didukung	Tipe array. Jika type diatur ke array.json atau arrays, data langsung diproses menggunakan serialisasi JSON. Jika type diatur ke array atau document.array, elemen-elemennya digabungkan menjadi string. Pemisah, yang ditentukan dalam properti splitter kolom, adalah koma (,) secara default.

  Penting

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

Tipe data Data Integration khusus: combine

Pemetaan tipe data MongoDB Reader

Tabel berikut mencantumkan pemetaan antara tipe data MongoDB dan tipe data Data Integration untuk MongoDB Reader.

Pemetaan tipe data MongoDB Writer

Contoh 1: Menggunakan tipe combine

Tipe data combine dari plugin MongoDB Reader memungkinkan Anda menggabungkan beberapa bidang dalam dokumen MongoDB menjadi satu string JSON. Misalnya, asumsikan Anda ingin mengimpor bidang dari tiga dokumen MongoDB ke MaxCompute. Dalam contoh berikut, bidang-bidang tersebut direpresentasikan oleh kunci, bukan pasangan kunci-nilai. Bidang a dan b umum di ketiga dokumen tersebut, sedangkan x_n adalah bidang variabel.

• doc1: a b x_1 x_2

• doc2: a b x_2 x_3 x_4

• doc3: a b x_5

Dalam file konfigurasi, Anda harus secara eksplisit menentukan bidang-bidang yang memerlukan pemetaan satu-ke-satu. Untuk bidang-bidang yang ingin Anda gabungkan, berikan nama baru yang berbeda dari nama bidang apa pun yang sudah ada dalam dokumen dan atur type-nya ke COMBINE. Kode berikut memberikan contohnya.

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

Tabel berikut menunjukkan output akhir di MaxCompute.

Contoh 2: Mengurai secara rekursif Document bersarang multi-level

Jika sebuah dokumen di MongoDB memiliki beberapa level nesting, Anda dapat mengonfigurasi tipe document untuk menguraikannya secara rekursif. Kode berikut memberikan 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"}

  Setelah konfigurasi selesai, bidang sumber dan tujuan dipetakan sebagai berikut: _id dipetakan ke id, name dipetakan ke name, dan a.b.c dipetakan ke c.

Dengan konfigurasi di atas, nilai bidang sumber bersarang a.b.c ditulis ke bidang tujuan c. Setelah tugas sinkronisasi dijalankan, data yang ditulis ke tujuan adalah this is value.

Konfigurasikan tugas sinkronisasi batch untuk satu tabel

• Untuk informasi lebih lanjut tentang prosedurnya, lihat Konfigurasikan tugas di Antarmuka tanpa kode dan Konfigurasikan tugas di editor kode.

• Untuk informasi lebih lanjut tentang semua parameter dan contoh skrip untuk editor kode, lihat Lampiran: Contoh skrip dan deskripsi parameter untuk MongoDB.

Konfigurasikan tugas sinkronisasi real-time untuk satu tabel

Lihat Konfigurasikan tugas sinkronisasi real-time untuk satu tabel.

Konfigurasikan tugas sinkronisasi untuk seluruh database

Anda dapat mengonfigurasi tugas untuk sinkronisasi batch, sinkronisasi real-time penuh dan inkremental, atau sinkronisasi real-time dari database sharded untuk seluruh database. Untuk informasi lebih lanjut, lihat Tugas sinkronisasi batch untuk seluruh database dan Konfigurasikan tugas sinkronisasi real-time untuk seluruh database.

Konfigurasikan 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 mode skrip. Informasi berikut menjelaskan parameter yang harus Anda konfigurasi untuk sumber data saat mengonfigurasi tugas sinkronisasi batch menggunakan editor kode.

Contoh skrip Reader

Skrip berikut adalah contoh pekerjaan yang dikonfigurasi untuk mengekstrak data dari MongoDB ke lingkungan lokal. Untuk informasi lebih lanjut tentang parameter, lihat deskripsi parameter 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": "", // Kueri penyaringan 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,// Menentukan apakah pembatasan kecepatan diaktifkan. Jika parameter ini diatur ke false, pembatasan kecepatan dinonaktifkan dan parameter mbps tidak berlaku. Jika diatur ke true, pembatasan kecepatan diaktifkan.
            "concurrent":1, // Jumlah pekerjaan konkuren.
            "mbps":"12"// Laju pembatasan. 1 mbps = 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Contoh skrip Writer

Skrip berikut adalah contoh pekerjaan sinkronisasi data yang dikonfigurasi untuk menulis data ke MongoDB. Untuk informasi lebih lanjut tentang parameter, lihat deskripsi parameter berikut.

{
    "type": "job",
    "version": "2.0",// Nomor versi.
    "steps": [
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "mongodb",// Nama plugin.
            "parameter": {
                "datasource": "",// Nama sumber data.
                "column": [
                    {
                        "name": "_id",// Nama kolom.
                        "type": "ObjectId"// Tipe data. Jika replaceKey adalah _id, Anda harus mengatur type ke ObjectId. Jika Anda mengatur type ke string, penggantian akan 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 koleksi.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {// Jumlah catatan error.
            "record": "0"
        },
        "speed": {
            "throttle": true,// Menentukan apakah pembatasan kecepatan diaktifkan. Jika parameter ini diatur ke false, pembatasan kecepatan dinonaktifkan dan parameter mbps tidak berlaku. Jika diatur ke true, pembatasan kecepatan diaktifkan.
            "concurrent": 1,// Jumlah pekerjaan konkuren.
            "mbps": "1"// Laju pembatasan. 1 mbps = 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 Anda menjalankan pekerjaan Data Integration, preSql yang dikonfigurasi dieksekusi terlebih dahulu. Fase penulisan data aktual baru dimulai setelah eksekusi preSql selesai. Parameter preSql tidak memengaruhi konten data yang ditulis. Parameter preSql menyediakan eksekusi idempoten untuk Data Integration. Misalnya, preSql Anda dapat digunakan untuk membersihkan data historis sebelum setiap eksekusi tugas berdasarkan aturan bisnis Anda. Dalam kasus ini, jika tugas gagal, Anda cukup menjalankan ulang pekerjaan Data Integration.

Persyaratan format untuk preSql adalah sebagai berikut:

• Anda harus mengonfigurasi field type untuk menentukan jenis operasi pra-eksekusi. Nilai yang didukung adalah drop dan remove. Contoh: "preSql":{"type":"remove"}.

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

  • remove: Menghapus data berdasarkan kondisi.

  • json: Anda dapat menggunakan objek JSON untuk menentukan kondisi penghapusan data. Contoh: "preSql":{"type":"remove", "json":"{'operationTime':{'$gte':ISODate('${last_day}T00:00:00.424+0800')}}"}. Dalam contoh ini, ${last_day} adalah parameter penjadwalan DataWorks dalam format $[yyyy-mm-dd]. Anda juga dapat menggunakan operator kondisional MongoDB lain 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 penghapusan data menggunakan API MongoDB standar berikut:

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

    Catatan

    Untuk menghapus data berdasarkan kondisi, kami menyarankan agar Anda menggunakan konfigurasi JSON.

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

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

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