人妻无码久久精品,欧美制服丝袜人妻另类,美女视频很黄很a免费

本文為您介紹數據攝入有關的source、sink、transform、route和pipeline模塊的開發參考。

支持的連接器

連接器	支持類型
連接器	Source	Sink
消息隊列Kafka	×	√
實時數倉Hologres	×	√
MySQL 說明支持連接RDS MySQL版、PolarDB MySQL版及自建MySQL。	√	×
Upsert Kafka	×	√
Print	×	√
StarRocks	×	√
流式數據湖倉Paimon	×	√

source模塊

source模塊定義數據攝入的數據源端，目前僅支持MySQL作為數據源。

語法結構

source:
  type: mysql
  name: mysql source
  xxx: ...

具體配置請查看對應連接器的數據攝入部分。

sink模塊

sink模塊定義數據攝入的目標端，目前支持的系統包括消息隊列Kafka、Upsert Kafka、實時數倉Hologres、流式數據湖倉Paimon、StarRocks和Print。

語法結構

sink:
  type: hologres
  name: hologres sink
  xxx: ...

具體配置請查看對應連接器的數據攝入部分。

transform模塊

您可以在YAML作業的transform語句塊中填寫若干規則信息，從而實現源表中數據的投影、計算和過濾等功能。

語法結構

transform:
  - source-table: db.tbl1
    projection: ...
    filter: ...
  - source-table: db.tbl2
    projection: ...
    filter: ...

配置項

參數	含義	是否必填	備注
`source-table`	指定生效上游表。	是	支持使用正則表達式。
`projection`	指定用于保留部分上游列的投影規則。	否	使用的句法與SQL SELECT語句類似。不填則不追加或刪除任何列。
`filter`	行過濾規則。	否	使用的句法與SQL WHERE語句類似。不填則不過濾任何行。
`primary-keys`	設定transform后Schema的主鍵列表。	否	不填則保留原Schema的主鍵定義。主鍵列表使用英文逗號（`,`）分隔。
`partition-keys`	設定transform的分區鍵列表。	否	不填則保留原Schema的分區鍵定義，分區鍵列表使用英文逗號（`,`）分隔。
`table-options`	需要傳遞給Sink的額外配置信息。	否	Options選項，例如Paimon Sink的分桶數、注釋等信息。
`description`	該transform塊的描述信息。	否	無。

計算列

您可以在projection規則中使用<Expression> AS <ColName>句法來定義計算列，表達式將對上游的每條數據分別求值后填入相應列。

警告

計算列的表達式不可以引用其他計算列的值，即使被引用的列出現在該計算列之前。例如a, b AS c, c AS d不是合法的projection表達式。

例如，在接收到來自上游db.tbl表的[+I, id = 1]數據記錄時，將其轉化為[+I, id = 1, inc_id = 2]數據行并發送給下游。

transform:
  - source-table: db.tbl
    projection: id, id + 1 AS inc_id

通配符

如果您希望將源表中的所有列以及后續追加的新列按原樣發送給下游，則可以在projection規則中使用星號（*）通配符。

說明

如果一個projection規則中沒有使用通配符（*），則其產生的Schema就是固定的，并且始終與projection規則中寫出的版本保持一致。

例如，*, 'extras' AS extras表示會在上游Schema的列尾追加額外的列，并持續將上游的表結構變更發送給下游。

transform:
  - source-table: db.tbl
    projection: \*, 'extras' AS extras

元數據列

在編寫projection規則時，可以將以下預先定義的元數據列作為普通數據列使用：

重要

請勿定義與元數據列同名的普通數據列。

元數據列名稱	數據類型	說明
`__namespace_name__`	String	這條數據變更記錄對應源表的Namespace名稱。
`__schema_name__`	String	這條數據變更記錄對應源表的Schema名稱。
`__table_name__`	String	這條數據變更記錄對應源表的Table名稱。
`__data_event_type__`	String	這條數據變更記錄對應的操作類型（`+I`、`-U`、`+U`、`-D`）。重要由于CDC Event總是將一次更新對應的Update Before和Update After打包為一條事件，因此`__data_event_type__`的內容在同一條Update事件里分別為`-U`和`+U`。請勿將其作為主鍵使用。

例如，將上游表的全限定名稱寫入計算列中，并發送給下游。

transform:
  - source-table: \.*.\.*
    projection: \*, __namespace_name__ || __schema_name__ || __table_name__ AS identifier

各個數據庫連接器對Namespace、Schema和Table名稱的映射關系如下表所示。

數據庫類型	Namespace名稱	Schema名稱	Table名稱
JDBC	Catalog	Schema	Table
Debezium	Catalog	Schema	Table
MySQL	Database	-	Table
Postgres	Database	Schema	Table
Oracle	-	Schema	Table
Microsoft SQL Server	Database	Schema	Table
StarRocks	Database	-	Table
Doris	Database	-	Table

注意事項

修改transform模塊的語句后，不能從已有的狀態恢復，需要進行無狀態啟動。
通常情況下，projection和filter語句無需使用引號包裹。
```
transform:
  - projection: a, b, c
    # 等價于
  - projection: "a, b, c"
```
然而，如果Projection表達式的第一個字符為*、'等特殊字符，則整行表達式可能無法被作為合法的YAML字符串字面量解析。此時需要手動使用引號包裹整個表達式，或是使用\轉義：
```
transform:
  - projection: *, 42      # 不是合法的YAML
  - projection: '*, 42'    # OK
  - projection: \*, 42     # OK  
```

route模塊

您可以在YAML作業的route模塊中定義包含若干條route規則的語句塊，描述上游表到下游表的復雜拓撲結構。

語法結構

route:
  - source-table: db.tbl1
    sink-table: sinkdb.tbl1
  - source-table: db.tbl2
    sink-table: sinkdb.tbl2

配置項

參數	含義	是否必填	備注
`source-table`	指定生效上游表。	是	支持使用正則表達式。
`sink-table`	指定數據路由的目標位置。	是	無。
`replace-symbol`	在使用模式匹配功能時，用于指代上游表名的字符串。	否	例如，當replace-symbol設置為`<>`時，可以將sink-table配置為`sinkdb.<>`。這樣，來自上游的表`table1`會被寫入到`sinkdb.table1`表中。
`description`	該route塊的描述信息。	否	無。

使用方法

一對一路由

將上游表mydb.web_order中的數據路由到下游表mydb.ods_web_order。

route:
  - source-table: mydb.web_order
    sink-table: mydb.ods_web_order
    description: sync table to one destination table with given prefix ods_

合并分庫分表

將上游mydb數據庫中的所有表合并到下游mydb.merged表中。

route:
  - source-table: mydb.\.*
    sink-table: mydb.merged
    description: sync sharding tables to one destination table

多路由規則

可以在一個route塊中使用YAML列表符號（-）定義多條規則，它們會同時生效。

route:
  - source-table: mydb.orders
    sink-table: ods_db.ods_orders
    description: sync orders table to orders
  - source-table: mydb.shipments
    sink-table: ods_db.ods_shipments
    description: sync shipments table to ods_shipments
  - source-table: mydb.products
    sink-table: ods_db.ods_products
    description: sync products table to ods_products

模式匹配

將source_db數據庫中的全部表一一對應地同步到sink_db中，并保持表名不變。

route:
  - source-table: source_db.\.*
    sink-table: sink_db.<>
    replace-symbol: <>
    description: route all tables in source_db to sink_db

使用replace-symbol參數定義的<>特殊字符串會被表名替代，從而實現源表到匯表的一一對應。

數據分發

將同一張表的數據分發給多個下游表，只需定義多條路由規則即可。例如將mydb.orders的數據會被同時分發到sink_db和backup_sink_db兩個數據庫中。

route:
  - source-table: mydb.orders
    sink-table: sink_db.orders
  - source-table: mydb.orders
    sink-table: backup_sink_db.orders

注意事項

修改route模塊的語句后，不能從已有的狀態恢復，需要進行無狀態啟動。

pipeline模塊

您可以在pipeline模塊配置數據攝入YAML作業的整體配置。

語法結構

pipeline:
  name: CDC YAML job
  schema.change.behavior: lenient

配置項

參數

說明

是否必填

數據類型

默認值

備注

name

數據攝入YAML名稱。

否

STRING

Flink CDC Pipeline Job

無。

schema.change.behavior

Schema變更行為配置。

否

STRING

lenient

可配置的值如下，詳見Schema變更行為配置。

lenient（默認值）
exception
evolve
try_evolve
ignore

Schema變更行為配置

數據攝入YAML作業支持將數據源的Schema變更同步到下游目標端，例如創建表、添加列、重命名列、更改列類型、刪除列和刪除表等。下游目標端可能不支持全部的Schema變更，您可以通過schema.change.behavior配置來修改Schema變更發生時目標端的處理方式。

Schema變更模式

模式	說明
lenient（默認）	數據攝入YAML作業會對Schema變更進行轉換成目標端可處理的變更并發送，遵循以下規則：不發送Drop table和Truncate table變更。列重命名時，改為發送更改列類型和新增列兩個事件。原有的列不刪除，更改列類型為nullable，同時新增一個列名為新名稱，數據類型改為nullable的列。刪除列時，改為發送更改列類型事件，將對應字段類型變為nullable。新增列時仍發送新增列事件，但字段類型會變為nullable。
exception	不允許任何Schema變更行為。當目標端不支持處理Schema變更時，可以使用此模式。收到Schema變更事件時，數據攝入YAML作業會拋出異常。
evolve	數據攝入YAML作業會將所有Schema更改應用于目標端。如果Schema變更在目標端應用失敗，數據攝入YAML作業會拋出異常并觸發故障重啟。
try_evolve	數據攝入YAML作業會嘗試將Schema變更應用到目標端，如果目標端不支持處理發送的Schema變更，數據攝入YAML作業不會失敗重啟，嘗試通過轉換后續數據方式進行處理。警告 try_evolve模式下，如果發生Schema變更應用失敗，可能導致上游后續到來的數據出現部分列丟失、被截斷等情況。
ignore	所有Schema變更都不會應用于目標端。當您的目標端尚未準備好進行任何Schema變更，想要繼續從未更改的列中接收數據時，可以使用此模式。

控制目標端接收的Schema變更

在某些場景下，不需要所有Schema變更同步到目標端。例如，允許新增列但禁止刪除列來避免刪除已有的數據。

您可以通過在sink模塊中設置include.schema.changes和exclude.schema.changes選項來控制。

參數	說明	是否必填	數據類型	默認值	備注
include.schema.changes	支持應用的Schema變更。	否	List<String>	無	默認支持所有變更。
exclude.schema.changes	不支持應用的Schema變更。	否	List<String>	無	優先級高于`include.schema.changes`。

以下是可配置架構變更事件類型的完整列表：

事件類型	說明
`add.column`	新增列。
`alter.column.type`	變更列類型。
`create.table`	創建表。
`drop.column`	刪除列。
`drop.table`	刪除表。
`rename.column`	修改列名。
`truncate.table`	清空數據。

說明

Schema變更支持部分匹配。例如，傳入drop相當于同時傳入drop.column和 drop.table。

代碼示例

示例1：Schema變更行為配置為evolve

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604

sink:
  type: values
  name: Values Sink
  print.enabled: true
  sink.print.logger: true
  
pipeline:
  name: mysql to print job
  schema.change.pipeline: evolve

示例2：支持創建表和列相關事件，不支持刪除列

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604

sink:
  type: values
  name: Values Sink
  print.enabled: true
  sink.print.logger: true
  include.schema.changes: [create.table, column] # 匹配了 CreateTable、AddColumn、AlterColumnType、RenameColumn、和 DropColumn 事件
  exclude.schema.changes: [drop.column] # 排除了 DropColumn 事件
  
pipeline:
  name: mysql to print job
  schema.change.pipeline: evolve

函數

內置函數

CDC YAML提供了豐富的內置函數，可以直接在transform模塊中的projection和filter表達式中使用。

比較函數

說明

除非特別說明，否則以下內置函數在輸入參數包含NULL時均返回NULL。

函數	說明
value1 = value2	如果value1等于value2，則返回TRUE；否則返回FALSE。
value1 <> value2	如果value1不等于value2，則返回TRUE；否則返回FALSE。
value1 > value2	如果value1大于value2，則返回TRUE；否則返回FALSE。
value1 >= value2	如果value1大于或等于value2，則返回TRUE；否則返回FALSE。
value1 < value2	如果value1小于value2，則返回TRUE；否則返回FALSE。
value1 <= value2	如果value1小于或等于value2，則返回TRUE；否則返回FALSE。
value IS NULL	如果value是NULL，則返回TRUE；否則返回FALSE。
value IS NOT NULL	如果value不是NULL，則返回TRUE；否則返回FALSE。
value1 BETWEEN value2 AND value3	如果value1的值介于value2和value3之間，則返回TRUE；否則返回FALSE。
value1 NOT BETWEEN value2 AND value3	如果value1的值并非介于value2和value3之間，則返回TRUE；否則返回FALSE。
string1 LIKE string2	如果string1的值與string2定義的模式匹配，則返回TRUE；否則返回FALSE。
string1 NOT LIKE string2	如果string1的值與string2定義的模式不匹配，則返回TRUE；否則返回FALSE。
value1 IN (value2 [, value3]* )	如果value1的值存在于[value2, value3, ...]列表中，則返回TRUE；否則返回FALSE。
value1 NOT IN (value2 [, value3]* )	如果value1的值不存在于[value2, value3, ...]列表中，則返回TRUE；否則返回FALSE。

邏輯函數

函數	說明
boolean1 OR boolean2	如果boolean1和boolean2至少有一個為TRUE，則返回TRUE。
boolean1 AND boolean2	如果boolean1和boolean2均為TRUE，則返回TRUE。
NOT boolean	如果boolean為TRUE，則返回FALSE；如果boolean是FALSE，則返回TRUE。
boolean IS FALSE	如果boolean為TRUE，則返回FALSE；如果boolean是FALSE，則返回TRUE。
boolean IS NOT FALSE	如果boolean為TRUE，則返回TRUE；如果boolean是FALSE，則返回FALSE。
boolean IS TRUE	如果boolean為TRUE，則返回TRUE；如果boolean是FALSE，則返回FALSE。
boolean IS NOT TRUE	如果boolean為TRUE，則返回FALSE；如果boolean是FALSE，則返回TRUE。

算數函數

函數	說明
numeric1 + numeric2	返回numeric1加上numeric2的值。
numeric1 - numeric2	返回numeric1減去numeric2的值。
numeric1 * numeric2	返回numeric1乘以numeric2的值。
numeric1 / numeric2	返回numeric1除以numeric2的值。
numeric1 % numeric2	返回numeric1對numeric2取模的值。
ABS(numeric)	返回numeric的絕對值。
CEIL(numeric)	返回numeric向上取整的值。
FLOOR(numeric)	返回numeric向下取整的值。
ROUND(numeric, int)	返回numeric四舍五入到小數點后n位的值。
UUID()	生成一個全局唯一ID（UUID）字符串（例如“3d3c68f7-f608-473f-b60c-b0c44ad4cc4e”）。使用RFC 4122 type 4方法偽隨機生成。

字符串函數

函數	說明
string1 \|\| string2	返回string1和string2拼接而成的字符串。重要請勿將其與邏輯或運算符混淆。
CHAR_LENGTH(string)	返回string字符串中的字符數。
UPPER(string)	返回string的大寫形式字符串。
LOWER(string)	返回string的小寫形式字符串。
TRIM(string1)	刪除string兩側的空白字符。
REGEXP_REPLACE(string1, string2, string3)	將string1中所有滿足string2模式的子串替換為string3。例如，`REGEXP_REPLACE('foobar', 'oo\|ar', '__')`求值的結果為 `f__b__`。
SUBSTRING(string FROM integer1 [ FOR integer2 ])	返回string從第integer1到第integer2個字符的子串。說明在不提供`FOR integer2`語句時，默認提取到字符串尾部。
CONCAT(string1, string2,…)	返回將string1、string2、…拼接在一起形成的新字符串。例如，`CONCAT('AA', 'BB', 'CC')`求值的結果為`AABBCC`。

時間函數

函數	說明
LOCALTIME	返回當前時區下的本地時間，返回類型為`TIME(0)`。
LOCALTIMESTAMP	返回當前時區下的本地時間戳，返回類型為`TIMESTAMP(3)`。
CURRENT_TIME	返回當前時區下的本地時間，與LOCAL_TIME相同。
CURRENT_DATE	返回當前時區下的本地日期。
CURRENT_TIMESTAMP	返回當前時區下的本地時間戳。返回類型為`TIMESTAMP_LTZ(3)`。
NOW()	返回當前時區下的本地時間戳，與CURRENT_TIMESTAMP相同。
DATE_FORMAT(timestamp, string)	將傳入的時間戳按指定的格式化字符串string進行格式化。說明格式化字符串與Java中的SimpleDateFormat格式兼容。
TIMESTAMPDIFF(timepointunit, timepoint1, timepoint2)	計算timepoint1和timepoint2之間差距多少timepointunit單位。 timepointunit可被指定為SECOND、MINUTE、HOUR、DAY、MONTH或YEAR。
TO_DATE(string1[, string2])	將傳入的日期字符串string1按string2指定的格式轉化為DATE類型。說明在不指定格式化字符串string2時，默認采用`yyyy-MM-dd`格式。
TO_TIMESTAMP(string1[, string2])	將傳入的時間戳字符串string1按string2指定的格式轉化為不帶時區信息的TIMESTAMP類型。說明在不指定格式化字符串string2時，默認采用`yyyy-MM-ddHH:mm:ss`格式。

說明

在進行projection和filter表達式求值時，可以保證其中每個子表達式所得到的時間點都一致。例如，NOW() AS t1, NOW() AS t2, NOW() AS t3得到的t1、t2、t3一定對應同一個時間戳，無論其求值時間和順序如何。

條件函數

函數	說明
CASE value WHEN value1_1 [, value1_2]* THEN RESULT1 (WHEN value2_1 [, value2_2 ]* THEN result_2)* (ELSE result_z) END	依次檢查value值是否等于WHEN子句給出的值，并返回第一個相等子句的RESULT值。如果沒有任何子句滿足條件，則返回ELSE子句指定的值。如果沒有指定ELSE子句，則返回NULL。
CASE WHEN condition1 THEN result1 (WHEN condition2 THEN result2)* (ELSE result_z) END	依次檢查value值是否滿足每個WHEN子句給出的條件，并返回第一個滿足條件子句的RESULT值。如果沒有任何子句滿足條件，則返回ELSE子句指定的值。如果沒有指定ELSE子句，則返回NULL。
COALESCE(value1 [, value2]*)	返回[value1、value2、……]列表中第一個不為NULL的元素。如果列表中所有元素均為NULL，則返回NULL。
IF(condition, true_value, false_value)	如果condition子句對應的條件為真，則返回true_value；否則返回false_value。

UDF函數

CDC YAML也支持使用Java語言編寫自定義UDF函數，并像內置函數一樣調用。

UDF函數定義

滿足以下要求的Java類可以作為CDC YAML UDF函數使用：

實現了org.apache.flink.cdc.common.udf.UserDefinedFunction接口。
擁有一個公共無參構造器。
至少含有一個名為eval的公共方法。

UDF函數類可以通過@Override以下接口來實現更精細的語義控制：

重寫getReturnType方法來手動指定方法的返回類型。
重寫open和close方法來插入生命周期函數。

例如，將傳入的整型參數增加1后返回的UDF函數定義如下。

public class AddOneFunctionClass implements UserDefinedFunction {
    
    public Object eval(Integer num) {
        return num + 1;
    }
    
    @Override
    public DataType getReturnType() {
        // 由于eval函數的返回類型不明確，需要
        // 使用getReturnType寫明確指定類型
        return DataTypes.INT();
    }
    
    @Override
    public void open() throws Exception {
        // ...
    }

    @Override
    public void close() throws Exception {
        // ...
    }
}

UDF函數注冊

通過在CDC YAMLpipeline塊中加入如下所示的定義即可注冊UDF函數：

pipeline:
  user-defined-function:
    - name: inc
      classpath: org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass
    - name: format
      classpath: org.apache.flink.cdc.udf.examples.java.FormatFunctionClass

說明

此處類路徑對應的JAR包需要作為外部依賴上傳。
UDF函數名稱可以在此處任意調整，無需與UDF類名一致。

UDF函數使用

在完成UDF函數注冊后，即可在projection和filter語句塊中，像內置函數一樣直接調用UDF函數。代碼示例如下。

transform:
  - source-table: db.\.*
    projection: "*, inc(inc(inc(id))) as inc_id, format(id, 'id -> %d') as formatted_id"
    filter: inc(id) < 100

Flink ScalarFucntion兼容性

繼承自ScalarFunction的Flink UDF函數也可以直接作為CDC YAML UDF函數注冊并使用，但存在以下限制：

不支持帶參數的ScalarFunction。
Flink風格的TypeInformation類型標注會被忽略。
open和close生命周期鉤子函數不會被調用。

日本熟妇hd丰满老熟妇,中文字幕一区二区三区在线不卡 ,亚洲成片在线观看,免费女同在线一区二区

支持的連接器

source模塊

語法結構

sink模塊

語法結構

transform模塊

語法結構

配置項

計算列

通配符

元數據列

注意事項

route模塊

語法結構

配置項

使用方法

一對一路由

合并分庫分表

多路由規則

模式匹配

數據分發

注意事項

pipeline模塊

語法結構

配置項

Schema變更行為配置

Schema變更模式

控制目標端接收的Schema變更

代碼示例

函數

內置函數

比較函數

邏輯函數

算數函數

字符串函數

時間函數

條件函數

UDF函數

UDF函數定義

UDF函數注冊

UDF函數使用

Flink ScalarFucntion兼容性

相關文檔