DataWorks：Sumber data DataHub

Versi DataHub yang Didukung

• DataHub Reader membaca data dari DataHub menggunakan DataHub SDK for Java. Berikut adalah contoh penggunaan DataHub SDK for Java:

  <dependency>
      <groupId>com.aliyun.DataHub</groupId>
      <artifactId>aliyun-sdk-DataHub</artifactId>
      <version>2.9.1</version>
  </dependency>

• DataHub Writer menulis data ke DataHub menggunakan DataHub SDK for Java. Berikut adalah contoh penggunaan DataHub SDK for Java:

  <dependency>
      <groupId>com.aliyun.datahub</groupId>
      <artifactId>aliyun-sdk-datahub</artifactId>
      <version>2.5.1</version>
  </dependency>

Batasan

Pembacaan dan penulisan data batch

String harus dikodekan dalam format UTF-8 dengan ukuran maksimum 1 MB per string.

Pembacaan dan penulisan data real-time

• Tugas sinkronisasi real-time mendukung kelompok sumber daya tanpa server dan kelompok sumber daya eksklusif.

• Ketika menyinkronkan data ke sumber data DataHub secara real-time, nilai hash dari data sumber diverifikasi. Data dengan nilai hash yang sama disinkronkan ke shard yang sama di sumber data DataHub.

Penulisan real-time data penuh dan inkremental

Setelah menjalankan solusi sinkronisasi, data penuh di sumber ditulis ke tujuan menggunakan tugas sinkronisasi batch. Kemudian, data inkremental di sumber ditulis ke tujuan menggunakan tugas sinkronisasi real-time. Saat menulis data ke DataHub, perhatikan hal-hal berikut:

• Anda hanya dapat menulis data ke topik bertipe TUPLE. Untuk informasi lebih lanjut tentang tipe data yang didukung oleh topik TUPLE, lihat Tipe Data.

• Ketika menjalankan tugas sinkronisasi real-time untuk menyinkronkan data ke DataHub secara real-time, lima bidang tambahan ditambahkan ke topik tujuan secara default. Anda juga dapat menambahkan bidang lain sesuai kebutuhan bisnis. Untuk informasi lebih lanjut tentang format pesan DataHub, lihat Lampiran: Format Pesan DataHub.

Pemetaan tipe data

Data disinkronkan berdasarkan pemetaan antara tipe data bidang di DataHub dan layanan tertentu. DataHub mendukung tipe data berikut: BIGINT, STRING, BOOLEAN, DOUBLE, TIMESTAMP, dan DECIMAL.

Tambahkan sumber data

Sebelum mengembangkan tugas sinkronisasi di DataWorks, tambahkan sumber data yang diperlukan ke DataWorks dengan mengikuti petunjuk di Manajemen Sumber Data. Anda dapat melihat tip informasi parameter di Konsol DataWorks untuk memahami arti parameter saat menambahkan sumber data.

Mengembangkan tugas sinkronisasi data

Untuk informasi tentang titik masuk dan prosedur konfigurasi tugas sinkronisasi, lihat panduan konfigurasi berikut.

Konfigurasikan tugas sinkronisasi batch untuk menyinkronkan data tabel tunggal

• Untuk informasi lebih lanjut tentang prosedur konfigurasi, lihat Konfigurasikan Tugas Sinkronisasi Batch Menggunakan Antarmuka Tanpa Kode dan Konfigurasikan Tugas Sinkronisasi Batch Menggunakan Editor Kode.

• Untuk informasi tentang semua parameter yang dikonfigurasi dan kode yang dijalankan ketika menggunakan editor kode untuk mengonfigurasi tugas sinkronisasi batch, lihat Lampiran: Kode dan Parameter.

Konfigurasikan tugas sinkronisasi real-time untuk menyinkronkan data tabel tunggal atau menyinkronkan semua data dari database

Konfigurasikan pengaturan sinkronisasi untuk menerapkan (real-time) sinkronisasi data penuh dan inkremental dalam tabel tunggal atau database

Untuk informasi lebih lanjut tentang prosedur konfigurasi, lihat Konfigurasikan Tugas Sinkronisasi di Data Integration.

FAQ

Apa yang harus saya lakukan jika data gagal ditulis ke DataHub karena jumlah data yang ingin saya tulis sekaligus melebihi batas maksimum?

Lampiran: Kode dan parameter

Lampiran: Konfigurasikan tugas sinkronisasi batch menggunakan editor kode

Kode untuk DataHub Reader

{
    "type":"job",
    "version":"2.0",// Nomor versi.
    "steps":[
        {
         "job": {
           "content": [
            {
                "reader": {
                    "name": "DataHubreader",
                    "parameter": {
                        "endpoint": "xxx" // Titik akhir DataHub.
                        "accessId": "xxx", // ID AccessKey yang digunakan untuk terhubung ke DataHub.
                        "accessKey": "xxx", // Rahasia AccessKey yang digunakan untuk terhubung ke DataHub.
                        "project": "xxx", // Nama proyek DataHub dari mana Anda ingin membaca data.
                        "topic": "xxx" // Nama topik DataHub dari mana Anda ingin membaca data.
                        "batchSize": 1000, // Jumlah catatan data yang dibaca sekaligus.
                        "beginDateTime": "20180910111214", // Waktu mulai konsumsi data.
                        "endDateTime": "20180910111614", // Waktu akhir konsumsi data.
                        "column": [
                            "col0",
                            "col1",
                            "col2",
                            "col3",
                            "col4"
                                  ]
                                }
                           },
                "writer": {
                    "name": "streamwriter",
                    "parameter": {
                        "print": false
                                 }
                            }
             }
           ]
         }
       }
     ],
    "setting":{
        "errorLimit":{
            "record":"0"// Jumlah maksimum catatan data kotor yang diizinkan.
        },
        "speed":{
            "throttle":true,// Menentukan apakah akan mengaktifkan pembatasan kecepatan. Nilai false menunjukkan bahwa pembatasan dinonaktifkan, dan nilai true menunjukkan bahwa pembatasan diaktifkan. Parameter mbps hanya berlaku ketika parameter throttle diatur ke true.
            "concurrent":1,// Jumlah maksimum thread paralel.
            "mbps":"12"// Laju transmisi maksimum. Unit: MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Parameter dalam kode untuk DataHub Reader

Kode untuk DataHub Writer

{
    "type": "job",
    "version": "2.0",// Nomor versi.
    "steps": [
        { 
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "datahub",// Nama plugin.
            "parameter": {
                "datasource": "",// Nama sumber data.
                "topic": "",// Unit minimum untuk langganan dan publikasi data. Anda dapat menggunakan topik untuk membedakan berbagai jenis data streaming.
                "maxRetryCount": 500,// Jumlah maksimum percobaan ulang jika tugas sinkronisasi gagal.
                "maxCommitSize": 1048576// Jumlah maksimum data yang di-buffer yang dapat diakumulasikan oleh Data Integration sebelum mengirimkan data ke tujuan. Unit: byte.
                 // DataHub memungkinkan hingga 10.000 catatan data ditulis dalam satu permintaan. Jika jumlah catatan data melebihi 10.000, tugas sinkronisasi gagal. Dalam hal ini, jumlah maksimum data yang dapat ditulis dalam satu permintaan dihitung menggunakan rumus berikut: Rata-rata jumlah data dalam satu catatan data × 10.000. Anda perlu mengatur maxCommitSize ke nilai yang lebih kecil dari jumlah data maksimum yang dihitung. Ini memastikan bahwa jumlah catatan data yang akan ditulis dalam satu permintaan tidak melebihi 10.000. Misalnya, jika ukuran data satu catatan data adalah 10 KB, nilai parameter ini harus kurang dari hasil 10 dikalikan 10.000.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": ""// Jumlah maksimum catatan data kotor yang diizinkan.
        },
        "speed": {
            "throttle":true,// Menentukan apakah akan mengaktifkan pembatasan kecepatan. Nilai false menunjukkan bahwa pembatasan dinonaktifkan, dan nilai true menunjukkan bahwa pembatasan diaktifkan. Parameter mbps hanya berlaku ketika parameter throttle diatur ke true.
            "concurrent":20, // Jumlah maksimum thread paralel.
            "mbps":"12"// Laju transmisi maksimum. Unit: MB/s.
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Parameter dalam kode untuk DataHub Writer