MaxCompute:Migrasikan RDS ke MaxCompute dengan partisi dinamis

Login ke DataWorks console.

Buat tabel tujuan di MaxCompute.

1. Di panel navigasi kiri, klik Workspace.

2. Di kolom Actions ruang kerja Anda, klik Quick Access > Data Development.

3. Klik kanan workflow yang telah Anda buat, lalu pilih New table > MaxCompute > Table.

4. Di halaman Create Table, pilih instans engine, skema, dan path. Masukkan Table Name, lalu klik Create.

5. Di halaman pengeditan tabel, klik ikon untuk beralih ke mode DDL.

6. Di kotak dialog DDL, masukkan pernyataan berikut, lalu klik Generate Table Schema. Di kotak dialog Confirm operation yang muncul, klik OK.

   CREATE TABLE IF NOT EXISTS ods_user_info_d (
   uid STRING COMMENT 'User ID',
   gender STRING COMMENT 'Gender',
   age_range STRING COMMENT 'Age range',
   zodiac STRING COMMENT 'Zodiac sign'
   )
   PARTITIONED BY (
   dt STRING
   );                           

7. Klik Submit to Production Environment.

Buat node sinkronisasi offline.

1. Buka halaman analitik data. Klik kanan workflow yang ditentukan, lalu pilih Create Node > Data Integration > Offline synchronization.

2. Di kotak dialog Create Node, masukkan Name, lalu klik Confirm.

3. Pilih sumber data, kelompok sumber daya, dan tujuan, lalu uji koneksi.

   • Source: Sumber data MySQL yang telah Anda buat.

   • Resource Group: Pilih grup sumber daya eksklusif untuk Integrasi Data.

   • Destination: Sumber data MaxCompute yang telah Anda buat.

4. Klik Next untuk mengonfigurasi tugas. Untuk sumber, pilih tabel ods_user_info_d dan atur split key ke uid. Untuk tujuan, pilih tabel ods_user_info_d dan masukkan dt=${bizdate} untuk informasi partisi. Atur Tunnel resource group ke public transport resource dan skema ke default. Untuk mode penulisan, pilih Insert Overwrite (Clear existing data before writing), dan untuk menulis string kosong sebagai null, pilih No.

Konfigurasi parameter penjadwalan.

1. Di panel navigasi kanan, klik Properties.

2. Di bagian Scheduling Parameter, parameter default adalah ${bizdate}, yang menggunakan format yyyymmdd.

   Catatan

   Nilai parameter default sesuai dengan Destination yang dikonfigurasi untuk Partition Information. Saat tugas dijalankan, sistem secara otomatis mengganti nilai partisi dengan tanggal bisnis (hari sebelum eksekusi tugas), karena pekerjaan ETL biasanya memproses data hari sebelumnya. Jika Anda perlu menggunakan tanggal eksekusi saat ini, Anda harus menyesuaikan parameter tersebut.

   Pengaturan parameter kustom: Anda dapat memilih tanggal dan format tertentu, seperti pada contoh berikut:

   • N tahun mendatang: $[add_months(yyyymmdd,12*N)]

   • N tahun sebelumnya: $[add_months(yyyymmdd,-12*N)]

   • N bulan sebelumnya: $[add_months(yyyymmdd,-N)]

   • N minggu mendatang: $[yyyymmdd+7*N]

   • N bulan mendatang: $[add_months(yyyymmdd,N)]

   • N minggu sebelumnya: $[yyyymmdd-7*N]

   • N hari mendatang: $[yyyymmdd+N]

   • N hari sebelumnya: $[yyyymmdd-N]

   • N jam mendatang: $[hh24miss+N/24]

   • N jam sebelumnya: $[hh24miss-N/24]

   • N menit mendatang: $[hh24miss+N/24/60]

   • N menit sebelumnya: $[hh24miss-N/24/60]

   Catatan

   • Gunakan tanda kurung siku ([]) untuk menentukan rumus perhitungan variabel kustom. Contohnya: key1=$[yyyy-mm-dd].

   • Satuannya secara default dalam hari untuk variabel kustom. Misalnya, $[hh24miss-N/24/60] merepresentasikan hasil dari (yyyymmddhh24miss - (N/24/60 * 1 hari)), diformat sebagai hh24miss.

   • Satuan perhitungan fungsi add_months adalah bulan. Misalnya, $[add_months(yyyymmdd,12*N)-M/24/60] merepresentasikan hasil dari (yyyymmddhh24miss - (12 * N * 1 bulan) - (M/24/60 * 1 hari)), diformat sebagai yyyymmdd.

Klik ikon untuk menjalankan kode.

Anda dapat melihat hasilnya di Runtime Log.

Filter data historis dari sumber ApsaraDB RDS berdasarkan tanggal.

Lakukan pengisian ulang data. Untuk informasi selengkapnya, lihat Kelola instans pengisian ulang data.

Periksa hasil ekstraksi data ApsaraDB RDS di log eksekusi.

Alibaba DI Console, Build 201805310000 .
Copyright 2018 Alibaba Group, All rights reserved .
Start Job[16961870], traceId [283789484710656#79023#None#None#228255635341196741#None#None#rds_sync], running in Pipeline[basecommon_
89484710656]
The Job[16961870] will run in PhysicsPipeline [basecommon_group_283789484710656_oxs] with requestId [4f44180d-300c-47c3-8ea3-805d2
2018-12-02 03:31:25 : ---
Reader: mysql
                column=["uid","gender","age_range","zodiac"]
                connection=[{"datasource":"xxx","table":["`ods_user_info_d`"]}]]
                where=[20180913                          ]
                splitPk=[uid                             ]
Writer: odps
                isCompress=[false                        ]
                partition=[dt=20180913                    ]
                truncate=[true                           ]
                datasource=[odps_first                   ]
                column=["uid","gender","age_range","zodiac"]
                emptyAsNull=[false                       ]
                table=[ods_user_info_d                    ]
Setting:
                errorLimit=[{"record":""}                 ]
                speed=[{"concurrent":1,"dmu":1,"mbps":"10","throttle":true}]
2018-12-02 03:31:26 : State: 1(SUBMIT) | Total: 0R 0B | Speed: 0R/s 0B/s | Error: 0R 0B | Stage: 0.0%
2018-12-02 03:31:36 : State: 3(RUN) | Total: 0R 0B | Speed: 0R/s 0B/s | Error: 0R 0B | Stage: 0.0%

Verifikasi hasilnya. Jalankan perintah berikut di klien MaxCompute untuk memeriksa apakah data telah ditulis.

SELECT count(*) from ods_user_info_d where dt = 20180913;

Sinkronkan semua data ke tabel sementara di MaxCompute dan buat node skrip SQL. Jalankan perintah berikut.

drop table if exists ods_user_t;
CREATE TABLE ods_user_t ( 
        dt STRING,
        uid STRING,
        gender STRING,
        age_range STRING,
        zodiac STRING);
--Simpan data dari tabel MaxCompute ke tabel sementara.
insert overwrite table ods_user_t select dt,uid,gender,age_range,zodiac from ods_user_info_d;         

Buat node tugas sinkronisasi bernama mysql_to_odps. Ini adalah tugas sederhana untuk menyinkronkan semua data dari RDS ke MaxCompute tanpa partisi.

Gunakan pernyataan SQL untuk melakukan partisi dinamis ke tabel target. Perintahnya sebagai berikut.

drop table if exists ods_user_d;
//Buat tabel ODPS terpartisi. Ini adalah tabel tujuan akhir.
    CREATE TABLE ods_user_d (
    uid STRING,
        gender STRING,
        age_range STRING,
        zodiac STRING
)
PARTITIONED BY (
    dt STRING
);
//Jalankan pernyataan SQL partisi dinamis. Ini secara otomatis mempartisi data berdasarkan bidang dt dari tabel sementara. Catatan dengan nilai yang sama di bidang dt ditempatkan di partisi yang dibuat untuk nilai tersebut.
//Misalnya, jika beberapa catatan memiliki nilai 20181025 di bidang dt, partisi dt=20181025 akan dibuat secara otomatis di tabel terpartisi ODPS.
//SQL partisi dinamis adalah sebagai berikut.
//Perhatikan bahwa bidang dt disertakan dalam pernyataan SELECT. Ini menentukan bahwa partisi dibuat secara otomatis berdasarkan bidang ini.
insert overwrite table ods_user_d partition(dt)select dt,uid,gender,age_range,zodiac from ods_user_t;
//Setelah impor selesai, hapus tabel sementara untuk menghemat biaya penyimpanan.
drop table if exists ods_user_t;

Konfigurasikan ketiga node tersebut ke dalam workflow untuk dijalankan secara berurutan.

Lihat proses eksekusi. Fokus pada proses partisi dinamis dari node terakhir.

=20181203065434115g3a2eqsa
Log view:
http://logview.odps.aliyun.com/logview/?h=http://service.odps.aliyun.com/api&p=DataWorks_DOC&i=20181203065434115g3a2eqsa&token=V0NaNDFxUmpn...
Summary:
resource cost: cpu 0.00 Core * Min, memory 0.00 GB * Min
inputs:
  dataworks_doc.ods_user_t: 20028 (119496 bytes)
outputs:
  dataworks_doc.ods_use xxx 20028 (119176 bytes)
Job run time: 0.000
Job run mode: service job
Job run engine: execution engine
M1:
  instance count: 1
  run time: 0.000
  instance time:
    min: 0.000, max: 0.000, avg: 0.000
  input records:
    ...

Verifikasi hasilnya. Jalankan perintah berikut di klien MaxCompute untuk memeriksa data yang telah ditulis.

SELECT count(*) from ods_user_d where dt = 20180913;