DataWorks：Sumber data MongoDB

Tipe data MongoDB yang didukung oleh MongoDB Reader

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

Saat 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 field 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 field dikonfigurasi sebagai tipe nested atau combine, field 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 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, field tersebut merupakan tipe nested. MongoDB Reader membaca properti Document berdasarkan path. Untuk contoh lengkap, 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 digabung 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 khusus Data Integration: 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 field dalam dokumen MongoDB menjadi satu string JSON. Misalnya, asumsikan Anda ingin mengimpor field dari tiga dokumen MongoDB ke MaxCompute. Dalam contoh berikut, field direpresentasikan oleh kunci, bukan pasangan kunci-nilai. Field a dan b umum di ketiga dokumen tersebut, sedangkan x_n adalah field 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 field yang memerlukan pemetaan satu-ke-satu. Untuk field yang ingin digabung, berikan nama baru yang berbeda dari nama field 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 bersarang, 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"}

  

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

Konfigurasikan task sinkronisasi batch untuk satu tabel

• Untuk informasi lebih lanjut tentang prosedurnya, lihat Konfigurasikan task di Antarmuka tanpa kode dan Konfigurasikan task 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 task sinkronisasi real-time untuk satu tabel

Untuk informasi lebih lanjut tentang prosedurnya, lihat Konfigurasikan task sinkronisasi real-time di Data Integration dan Konfigurasikan task sinkronisasi real-time di DataStudio.

Konfigurasikan task sinkronisasi untuk seluruh database

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

Konfigurasikan task sinkronisasi batch menggunakan editor kode

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

Contoh skrip Reader

Skrip berikut adalah contoh job 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 field.
                        "type": "string" // Tipe field.
                    },
                    {
                        "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 pengendalian aliran diaktifkan. Jika parameter ini diatur ke false, pengendalian aliran dinonaktifkan dan parameter mbps tidak berlaku. Jika diatur ke true, pengendalian aliran diaktifkan.
            "concurrent":1, // Jumlah job konkuren.
            "mbps":"12"// Laju pengendalian aliran. 1 mbps = 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Contoh skrip Writer

Skrip berikut adalah contoh job 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 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 pengendalian aliran diaktifkan. Jika parameter ini diatur ke false, pengendalian aliran dinonaktifkan dan parameter mbps tidak berlaku. Jika diatur ke true, pengendalian aliran diaktifkan.
            "concurrent": 1,// Jumlah job konkuren.
            "mbps": "1"// Laju pengendalian aliran. 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 job 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 task berdasarkan aturan bisnis Anda. Dalam kasus ini, jika task gagal, Anda cukup menjalankan ulang job Data Integration.

Persyaratan format untuk preSql adalah sebagai berikut:

• Anda harus mengonfigurasi field type untuk menentukan jenis operasi awal. 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 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 kueri 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.