Realtime Compute for Apache Flink:Flink CDC のソースモジュールと sink モジュール

構文

source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog
  
sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog

ソースおよび sink モジュールの using.built-in-catalog パラメーターを使用して、既存の組み込みカタログを参照できます。

例えば、上記の例では、mysql_rds_catalog のカタログメタデータには、hostname、username、password などの必須パラメーターがすでに含まれています。そのため、これらのパラメーターを YAML ジョブで再度指定する必要はありません。

注意事項

以下のコネクタは、カタログからの接続情報の再利用をサポートしています：

• MySQL (ソース)

• Kafka (ソース)

• Upsert Kafka (sink)

• StarRocks (sink)

• Hologres (sink)

• Paimon (sink)

• Simple Log Service (ソース)

• Iceberg (sink)

構文

source:                                                                                                                                                                                                                                                                                                          
  type: mysql                                                                                                                                                                                                                                                                                                     
  host: localhost                                                                                                                                                                                                                                                                                                  
  port: 3306                                                                                                                                                                                                                                                                                                       
  username: admin
  password: pass
  tables: mydb.orders
  source-expand:
  # mydb.orders を 3 つのテーブルに複製し、それぞれを sink に書き込む前に異なる方法で処理します。
    - input-table: mydb.orders
      output-table: [ dwd.orders_full, dws.orders_summary, ads.orders_report ]

# 拡張された 3 つのテーブルそれぞれに異なる変換を適用します
transform:
  # DWD レイヤー：すべてのフィールドを保持し、計算列を追加します
  - source-table: dwd.orders_full
    projection: "*, amount * discount as final_price"
  # DWS レイヤー：キーフィールドのみを保持し、少額の注文をフィルターします
  - source-table: dws.orders_summary
    projection: order_id, user_id, amount, order_status
    filter: amount > 100
    primary-keys: order_id
  # ADS レイヤー：レポートに必要なフィールドのみを保持し、キャンセルされた注文をフィルターします
  - source-table: ads.orders_report
    projection: order_id, user_id, amount, TO_UPPER(order_status) as status
    filter: order_status <> 'CANCELLED'
    primary-keys: order_id

# 拡張された 3 つのテーブルを異なる sink テーブルにルーティングします
route:
  - source-table: dwd.orders_full
    sink-table: starrocks_dwd.orders_full_detail
  - source-table: dws.orders_summary
    sink-table: starrocks_dws.orders_summary
  - source-table: ads.orders_report
    sink-table: starrocks_ads.orders_report

sink:
  type: starrocks
  name: sink-starrocks
  jdbc-url: jdbc:mysql://localhost:9030
  load-url: localhost:8030
  username: root
  password: pass

例に示すように、この設定は mydb.orders テーブルのデータを dwd.orders_full、dws.orders_summary、ads.orders_report の 3 つの論理テーブルに複製します。これらの各テーブルは、個別の sink テーブルに書き込まれる前に、それぞれ異なる方法で処理されます。

注意事項

• VVR 11.6 以降で利用可能です。

• デフォルトでは、データ分散後に元のテーブルは保持されません。構文の例では、mydb.orders テーブルはデータストリームから削除されます。元のテーブルを保持するには、output-table リストに追加します。

  source-expand:
    - input-table: mydb.orders
      output-table: [ mydb.orders, db1.orders, db2.orders ]

• input-table および output-table パラメーターは正規表現をサポートしていません。