MaxCompute:Dinamik bölümleme ile RDS verilerini MaxCompute'a geçirin

Ön koşullar

• Bir DataWorks ortamı ayarlayın.

  1. MaxCompute'u etkinleştirin.

  2. DataWorks'te bir iş akışı oluşturun. Bu eğitimde DataWorks'ün temel modu kullanılır. Daha fazla bilgi için bkz. İş akışı oluşturma.

• Veri kaynaklarını ekleyin.

  • Kaynak olarak bir MySQL veri kaynağı ekleyin. Daha fazla bilgi için bkz. MySQL veri kaynağını yapılandırma.

  • RDS verilerini almak üzere hedef veri kaynağı olarak bir MaxCompute veri kaynağı ekleyin. Daha fazla bilgi için bkz. MaxCompute veri kaynağını yapılandırma.

Bölümleri otomatik olarak oluşturun

ApsaraDB RDS'den MaxCompute'e veri senkronizasyonu yapan ve tarih tabanlı bölümleri otomatik olarak oluşturan günlük bir görev zamanlayın. Veri senkronizasyon görevleri hakkında daha fazla bilgi için bkz. Veri Geliştirme ve İşletme.

1. DataWorks konsolu adresine gidin.

2. MaxCompute'te bir hedef tablo oluşturun.

   1. Sol gezinme bölmesinde, Workspace öğesine tıklayın.

   2. Çalışma alanınızın Actions sütununda, Quick Access > Data Development seçeneklerini belirleyin.

   3. Oluşturduğunuz iş akışına sağ tıklayın ve New table > MaxCompute > Table yolunu izleyin.

   4. Create Table sayfasında bir motor örneği, şema ve yol seçin. Bir Table Name girin ve Create öğesine tıklayın.

   5. Tablo düzenleme sayfasında, DDL moduna geçmek için simgesine tıklayın.

   6. DDL iletişim kutusunda aşağıdaki ifadeyi girin ve Generate Table Schema öğesine tıklayın. Açılan Confirm operation iletişim kutusunda OK seçeneğine tıklayın.

      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. Submit to Production Environment seçeneğine tıklayın.

3. Bir çevrimdışı senkronizasyon düğümü oluşturun.

   1. Veri analizi sayfasına gidin. Belirtilen iş akışına sağ tıklayın ve Create Node > Data Integration > Offline synchronization yolunu izleyin.

   2. Create Node iletişim kutusunda bir Name girin ve Confirm seçeneğine tıklayın.

   3. Veri kaynağını, kaynak grubunu ve hedefi seçin, ardından bağlantıyı test edin.

      • Source: Oluşturduğunuz MySQL veri kaynağı.

      • Resource Group: Data Integration için özel bir kaynak grubu seçin.

      • Destination: Oluşturduğunuz MaxCompute veri kaynağı.

   4. Görevi yapılandırmak için Next öğesine tıklayın. Kaynak için ods_user_info_d tablosunu seçin ve bölme anahtarını uid olarak ayarlayın. Hedef için ods_user_info_d tablosunu seçin ve bölüm bilgisi olarak dt=${bizdate} değerini girin. Tunnel kaynak grubunu public transport resource, şemayı ise default olarak belirleyin. Yazma modu için Insert Overwrite (Clear existing data before writing) seçeneğini işaretleyin, boş dizelerin null olarak yazılması konusunda ise No seçimini yapın.

4. Zamanlama parametrelerini yapılandırın.

   1. Sağ gezinme bölmesinde Properties seçeneğine tıklayın.

   2. Scheduling Parameter bölümünde varsayılan parametre ${bizdate} olup yyyymmdd biçimini kullanır.

      Not

      Varsayılan parametre değeri, Partition Information için yapılandırılan Destination ile eşleşir. ETL işleri genellikle önceki günün verilerini işlediğinden sistem, bölüm değerini görevin çalıştırıldığı tarihten bir gün önceki iş tarihiyle otomatik olarak değiştirir. Geçerli çalışma tarihini kullanmak istiyorsanız parametreyi özelleştirin.

      Tarih biçimlerini aşağıdaki ifadelerle özelleştirebilirsiniz:

      • N yıl sonra: $[add_months(yyyymmdd,12*N)]

      • N yıl önce: $[add_months(yyyymmdd,-12*N)]

      • N ay önce: $[add_months(yyyymmdd,-N)]

      • N hafta sonra: $[yyyymmdd+7*N]

      • N ay sonra: $[add_months(yyyymmdd,N)]

      • N hafta önce: $[yyyymmdd-7*N]

      • N gün sonra: $[yyyymmdd+N]

      • N gün önce: $[yyyymmdd-N]

      • N saat sonra: $[hh24miss+N/24]

      • N saat önce: $[hh24miss-N/24]

      • N dakika sonra: $[hh24miss+N/24/60]

      • N dakika önce: $[hh24miss-N/24/60]

      Not

      • Özel bir değişkenin hesaplama formülünü tanımlamak için köşeli parantezler ([]) kullanın. Örneğin: key1=$[yyyy-mm-dd].

      • Varsayılan olarak özel değişkenlerin hesaplama birimi gündür. Örneğin, $[hh24miss-N/24/60] ifadesi, (yyyymmddhh24miss - (N/24/60 * 1 day)) işleminin sonucunu hh24miss biçiminde temsil eder.

      • add_months işlevinin hesaplama birimi aydır. Örneğin, $[add_months(yyyymmdd,12*N)-M/24/60] ifadesi, (yyyymmddhh24miss - (12 * N * 1 month) - (M/24/60 * 1 day)) işleminin sonucunu yyyymmdd biçiminde temsil eder.

5. Kodu çalıştırmak için simgesini tıklayın.

6. Sonuçları görüntülemek için Runtime Log sekmesini kullanabilirsiniz.

Geçmiş verileri doldurma

Zamanlanmış görevden önceki geçmiş verileri eşitlemek için, verileri eşitlemek ve bölümleri otomatik olarak oluşturmak amacıyla DataWorks Operation Center içindeki Complement data özelliğini kullanın.

1. ApsaraDB RDS kaynağındaki geçmiş verileri tarihe göre filtreleyin.

   Eşitleme düğümünün Source bölümünde, Data Filter koşulunu ${bizdate} olarak ayarlayın.

2. Veri doldurma işlemi gerçekleştirin. Daha fazla bilgi için bkz. Manage data backfill instances.

3. Çalıştırma günlüğünde ApsaraDB RDS verilerinin çıkarma sonuçlarını kontrol edin.

   Çalıştırma günlüğü, MaxCompute'ün bölümü otomatik olarak oluşturduğunu gösterir. Okuyucu yapılandırmasındaki where yan tümcesi (where=[20180913]), yazıcı yapılandırmasındaki partition ile eşleşir (partition=[dt=20180913]

   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%

4. Sonucu doğrulayın. Verilerin yazılıp yazılmadığını kontrol etmek için MaxCompute istemcisinde aşağıdaki komutu çalıştırın.

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

Tarih dışı alanların karma kullanılarak bölümlenmesi

Verileri eyalet gibi tarih dışı bir alana göre bölümlediğinizde, Data Integration otomatik bölümleme gerçekleştiremez. Bunun yerine, RDS'deki bir alanı karmalayarak aynı alan değerine sahip kayıtları ilgili MaxCompute bölümünde saklayabilirsiniz.

1. Tüm verileri MaxCompute'daki geçici bir tabloya eşitleyin ve bir SQL betiği düğümü oluşturun. Aşağıdaki komutu çalıştırın.

   drop table if exists ods_user_t;
   CREATE TABLE ods_user_t ( 
           dt STRING,
           uid STRING,
           gender STRING,
           age_range STRING,
           zodiac STRING);
   --Store the data from the MaxCompute table in the temporary table.
   insert overwrite table ods_user_t select dt,uid,gender,age_range,zodiac from ods_user_info_d;         

2. mysql_to_odps adlı bir eşitleme görevi düğümü oluşturun. Bu, tüm verileri RDS'den MaxCompute'a eşitlemek için basit bir görevdir. Bölümler gerekmez.

3. Hedef tabloya dinamik bölümleme yapmak için bir SQL ifadesi kullanın. Komut aşağıdaki gibidir.

   drop table if exists ods_user_d;
   //Create a partitioned ODPS table. This is the final destination table.
       CREATE TABLE ods_user_d (
       uid STRING,
           gender STRING,
           age_range STRING,
           zodiac STRING
   )
   PARTITIONED BY (
       dt STRING
   );
   //Run the dynamic partitioning SQL statement. It automatically partitions data based on the dt field of the temporary table. Records with the same value in the dt field are placed in a partition created for that value.
   //For example, if some records have the value 20181025 in the dt field, a partition dt=20181025 is automatically created in the ODPS partitioned table.
   //The dynamic partitioning SQL is as follows.
   //Note that the dt field is included in the SELECT statement. This specifies that partitions are automatically created based on this field.
   insert overwrite table ods_user_d partition(dt)select dt,uid,gender,age_range,zodiac from ods_user_t;
   //After the import is complete, delete the temporary table to save storage costs.
   drop table if exists ods_user_t;

   MaxCompute'da verileri eşitlemek için SQL ifadelerini kullanabilirsiniz.

4. Üç düğümü sırayla çalışacak şekilde bir iş akışında yapılandırın.

   Çalıştırma sırası geçici tablo, mysql_to_odps ve hedef tablo şeklindedir.

5. Çalıştırma sürecini görüntüleyin. Son düğümün dinamik bölümleme sürecine odaklanın.

   =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:
       ...

6. Sonucu doğrulayın. Yazılan verileri kontrol etmek için MaxCompute istemcisinde aşağıdaki komutu çalıştırın.

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