Elasticsearch:Sinkronisasi data MaxCompute ke Alibaba Cloud Elasticsearch menggunakan DataWorks

Langkah 1: Siapkan data sumber

Buat tabel MaxCompute dan impor data uji ke dalam tabel tersebut. Untuk informasi lebih lanjut, lihat Buat tabel dan Impor data.

Topik ini menggunakan skema tabel dan data sampel berikut:

• Skema tabel

  Tabel berisi 7 bidang dan 1 bidang partisi. Bidang-bidang tersebut didefinisikan sebagai berikut:

  • create_time (string, primary key)

  • category (string)

  • brand (string)

  • buyer_id (string)

  • trans_num (bigint)

  • trans_amount (double)

  • click_cnt (bigint)

  Bidang partisi adalah pt (bigint).

• Data sampel

  Tabel sumber berisi bidang-bidang berikut:

  • create_time: tanggal transaksi, misalnya, 2020/6/1

  • category: kategori produk, misalnya, Outerwear, Fresh Food, Electronics, Bathroom

  • brand: nama merek, misalnya, Brand A hingga Brand G

  • buyer_id: ID pembeli, misalnya, user1 hingga user13

  • trans_num: jumlah transaksi

  • trans_amount: jumlah transaksi

  • click_cnt: jumlah klik

  • pt: bidang partisi, dengan nilai 1

Langkah 2: Beli dan konfigurasi grup sumber daya eksklusif

Beli grup sumber daya eksklusif untuk Integrasi Data dan asosiasikan dengan VPC serta ruang kerja. Grup sumber daya eksklusif memastikan transmisi data yang cepat dan stabil.

1. Masuk ke Konsol DataWorks .

2. Di bilah navigasi atas, pilih wilayah. Di panel navigasi kiri, klik Resource Groups.

3. Pada tab Exclusive Resource Group, pilih Create Resource Group of Old Version > Resource Group for Data Integration.

4. Pada halaman pembelian DataWorks Exclusive Resources (Subscription), atur Exclusive Resource Type menjadi Exclusive Resources for Data Integration, masukkan nama untuk grup sumber daya, lalu klik Buy Now.

   Untuk informasi lebih lanjut tentang konfigurasi, lihat Langkah 1: Beli grup sumber daya.

5. Temukan grup sumber daya eksklusif yang telah Anda buat dan klik Network Settings di kolom Actions untuk mengasosiasikannya dengan VPC. Untuk informasi lebih lanjut, lihat Bind a VPC.

   Catatan

   Topik ini menggunakan grup sumber daya eksklusif untuk Integrasi Data guna menyinkronkan data melalui VPC. Untuk informasi tentang cara menyinkronkan data melalui internet, lihat Configure an allowlist.

   Untuk mengaktifkan sinkronisasi data, grup sumber daya eksklusif harus terhubung ke VPC tempat kluster Alibaba Cloud Elasticsearch berada. Oleh karena itu, Anda harus mengasosiasikan grup sumber daya eksklusif dengan Virtual Private Cloud (VPC), Zone, dan VSwitch kluster Alibaba Cloud Elasticsearch Anda. Untuk melihat informasi ini, lihat View the basic information of an Elasticsearch cluster.

   Penting

   Setelah mengasosiasikan grup sumber daya eksklusif dengan VPC, Anda harus menambahkan vSwitch CIDR Block ke daftar izin IP privat VPC kluster Alibaba Cloud Elasticsearch. Untuk informasi lebih lanjut, lihat Configure a public or private IP address allowlist for an Elasticsearch cluster.

6. Di pojok kiri atas halaman, klik ikon kembali untuk kembali ke halaman Resource List.

7. Temukan grup sumber daya eksklusif yang telah Anda buat dan klik Associate Workspace di kolom Actions untuk mengasosiasikannya dengan ruang kerja target.

   Untuk informasi lebih lanjut, lihat Langkah 2: Asosiasikan ruang kerja.

Langkah 3: Tambahkan sumber data

Tambahkan MaxCompute dan Alibaba Cloud Elasticsearch sebagai sumber data di Integrasi Data.

1. Buka halaman Data Integration.

   1. Masuk ke Konsol DataWorks .

   2. Di panel navigasi kiri, klik Workspace.

   3. Temukan ruang kerja target dan pilih Shortcuts > Data Integration di kolom Actions.

2. Di panel navigasi kiri, klik Data Source.

3. Tambahkan sumber data MaxCompute.

   1. Pada halaman Data Sources, klik Add Data Source.

   2. Pada halaman Add Data Source, temukan dan pilih tipe sumber data MaxCompute.

   3. Pada kotak dialog Add MaxCompute data source, konfigurasikan parameter pada bagian Basic Information.

      Untuk informasi selengkapnya, lihat Mengonfigurasi sumber data MaxCompute.

   4. Pada bagian Connection Configuration, klik Test Connectivity. Status Connected menunjukkan bahwa koneksi berhasil.

   5. Klik Complete.

4. Ikuti prosedur yang sama untuk menambahkan sumber data Elasticsearch. Untuk informasi lebih lanjut, lihat Configure an Elasticsearch data source.

Langkah 4: Konfigurasi dan jalankan task sinkronisasi batch

Task sinkronisasi batch dijalankan menggunakan grup sumber daya eksklusif. Grup sumber daya mengambil data dari sumber data di Integrasi Data dan menulis data ke kluster Alibaba Cloud Elasticsearch.

1. Buka halaman Data development di DataWorks.

   1. Masuk ke Konsol DataWorks .

   2. Di panel navigasi kiri, klik Workspace.

   3. Temukan ruang kerja target dan pilih Shortcuts > Data development di kolom Actions.

2. Buat node sinkronisasi offline.

   1. Pada tab Data Development (), pilih Create > New business process.

   2. Klik kanan alur kerja yang telah Anda buat dan pilih Create Node > Data Integration > Batch Synchronization.

   3. Pada kotak dialog Create Node, masukkan nama untuk node tersebut dan klik OK.

3. Konfigurasi jaringan dan grup sumber daya.

   1. Pada bagian Source, atur Source menjadi MaxCompute(ODPS) dan pilih sumber data Anda di bawah Data Source Name.

   2. Pada bagian Resource Group, pilih grup sumber daya eksklusif.

   3. Pada bagian Destination, atur Destination menjadi Elasticsearch dan pilih sumber data tujuan Anda di bawah Data Source Name.

4. Klik Next step.

5. Konfigurasi task.

   1. Pada bagian Source, pilih tabel sumber.

   2. Pada bagian Destination, konfigurasikan parameter untuk tujuan.

   3. Pada bagian Field Mapping, petakan Source Column ke Target Column.

   4. Pada bagian Channel Control, konfigurasikan parameter channel.

   Untuk informasi selengkapnya tentang konfigurasi, lihat Mengonfigurasi Tugas Sinkronisasi dalam Mode Wizard.

6. Jalankan task.

   1. (Opsional) Konfigurasi properti penjadwalan untuk task. Di panel kanan, klik Properties dan konfigurasikan parameter penjadwalan sesuai kebutuhan. Untuk informasi lebih lanjut tentang parameter, lihat Scheduling configuration.

   2. Di pojok kanan atas area node, klik ikon Simpan untuk menyimpan task.

   3. Di pojok kanan atas area node, klik ikon Commit.

      Jika Anda mengonfigurasi properti penjadwalan untuk task, task akan berjalan secara otomatis pada interval yang dijadwalkan. Anda juga dapat mengklik ikon Run di pojok kanan atas area node untuk menjalankan task segera.

      Pesan Shell run successfully! dalam log eksekusi menunjukkan bahwa task berhasil dijalankan. Contoh log eksekusi task adalah sebagai berikut:

      2023-10-31 16:52:35 INFO Exit code of the Shell command 0
      2023-10-31 16:52:35 INFO --- Invocation of Shell command completed ---
      2023-10-31 16:52:35 INFO Shell run successfully!
      2023-10-31 16:52:35 INFO Current task status: FINISH
      2023-10-31 16:52:35 INFO Cost time is: 33.106s

Langkah 5: Verifikasi hasil sinkronisasi data

Di konsol Kibana, lihat data yang telah disinkronkan dan jalankan kueri.

1. Masuk ke konsol Kibana kluster Alibaba Cloud Elasticsearch target.

   Untuk informasi lebih lanjut, lihat Log on to the Kibana console.

2. Di pojok kiri atas halaman Kibana, klik ikon dan pilih Dev Tools.

3. Di Console, jalankan perintah berikut untuk melihat data yang telah disinkronkan.

   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {}}
   }

   Catatan

   odps_index adalah nilai bidang index yang Anda atur dalam skrip sinkronisasi data.

   Sinkronisasi yang berhasil mengembalikan hasil berikut.

   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {}}
   }
   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "_source": ["category", "brand"]
   }
   POST /odps_index/_search?pretty
   {
     "query": { "match": {"category":"生鲜"} }
   }
   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "sort": { "trans_num": { "order": "desc" }
           }
   }
   
   --- Response ---
   {
     "took" : 2,
     "timed_out" : false,
     "_shards" : {
       "total" : 1,
       "successful" : 1,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : 13,
       "max_score" : null,
       "hits" : [
         {
           "_index" : "odps_index",
           "_type" : "_doc",
           "_id" : "2020/6/7 8:00",
           "_score" : null,
           "_source" : {
             "trans_num" : 88,
             "click_cnt" : 80,
             "category" : "外套",
             "buyer_id" : "user7",
             "trans_amount" : 150.0,
             "brand" : "品牌E"
           },
           "sort" : [
             88
           ]
         },
         {
           "_index" : "odps_index",
           "_type" : "_doc",
           "_id" : "2020/6/11 8:00",
           "_score" : null,
           "_source" : {
             "trans_num" : 22,
             "click_cnt" : 70,
             "category" : "卫浴",
             "buyer_id" : "user11",
             "trans_amount" : 4500.0,
             "brand" : "品牌G"
           },
           "sort" : [
             22
           ]
         }
       ]
     }
   }

4. Jalankan perintah berikut untuk mencari bidang category dan brand dalam dokumen.

   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "_source": ["category", "brand"]
   }

5. Jalankan perintah berikut untuk mencari dokumen di mana category adalah fresh food.

   POST /odps_index/_search?pretty
   {
   "query": { "match": {"category":"生鲜"} }
   }

6. Jalankan perintah berikut untuk mengurutkan dokumen berdasarkan bidang trans_num.

   POST /odps_index/_search?pretty
   {
   "query": { "match_all": {} },
   "sort": { "trans_num": { "order": "desc" } }
   }

   Untuk perintah dan metode akses lainnya, lihat Elastic.co Help Center.