Delta Lake 在流利说数据接入中的架构和实践

CREATE TABLE `kafka_{db_name}_{table_name}` (`key` BINARY, `value` BINARY, `topic` STRING, `partition` INT, `offset` BIGINT, `timestamp` TIMESTAMP, `timestampType` INT)USING kafkaOPTIONS (  `kafka.sasl.mechanism` 'PLAIN',  `subscribe` 'cdc-{db_name}-{table_name}',  `serialization.format` '1',  `kafka.sasl.jaas.config` '*********(redacted)',  `kafka.bootstrap.servers` '{bootstrap-servers}',  `kafka.security.protocol` 'SASL_PLAINTEXT')

{  "payload": {    "before": {        db记录变更前的schema及内容，op=c时，为null    },    "after": {       db记录变更后的schema及内容，op=d时，为null    },    "source": {       ebezium配置信息    },    "op": "c",    "ts_ms":  }}

CREATE TABLE IF NOT EXISTS delta.delta_{db_name}_{table_name}(    {row_key_info},    ts_ms bigint,    json_record string,    operation_type string,    offset bigint)USING deltaLOCATION '------/delta/{db_name}.db/{table_name}'

CREATE SCAN incremental_{db_name}_{table_name} on kafka_{db_name}_{table_name} USING STREAMOPTIONS(startingOffsets='earliest',maxOffsetsPerTrigger='1000000',failOnDataLoss=false);CREATE STREAM jobOPTIONS(checkpointLocation='------/delta/{db_name}.db/{table_name}_checkpoint',triggerIntervalMs='300000')MERGE INTO delta.delta_{db_name}_{table_name} as targetUSING  (SELECT *  FROM (SELECT ts_ms, offset, operation_type, {key_column_sql}, coalesce(after_record, before_record) as after_record, row_number() OVER (PARTITION BY {key_column_partition_sql} ORDER BY ts_ms DESC, offset DESC) as rankFROM (SELECT ts_ms, offset, operation_type, before_record, after_record, {key_column_include_sql}FROM ( SELECT get_json_object(string(value), '$.payload.op') as operation_type,get_json_object(string(value), '$.payload.before') as before_record,get_json_object(string(value), '$.payload.after') as after_record,get_json_object(string(value), '$.payload.ts_ms') as ts_ms,offsetFROM incremental_{db_name}_{table_name}) binlog) binlog_wo_init ) binlog_rank where rank = 1) as sourceON {key_column_condition_sql}WHEN MATCHED AND (source.operation_type = 'u' or source.operation_type='d') THENUPDATE SET {set_key_column_sql}, ts_ms=source.ts_ms, json_record=source.after_record, operation_type=source.operation_type, offset=source.offsetWHEN NOT MATCHED AND (source.operation_type='c' or source.operation_type='u' or source.operation_type='d') THENINSERT ({inser_key_column_sql}, ts_ms, json_record, operation_type, offset) values ({insert_key_column_value_sql}, source.ts_ms, source.after_record, source.operation_type, source.offset);

optimize delta_{db_name}_{table_name};set spark.databricks.delta.retentionDurationCheck.enabled = false;VACUUM delta_{db_name}_{table_name} RETAIN 1 HOURS;

CREATE DATABASE IF NOT EXISTS {db_name} LOCATION '------/delta/{db_name}.db';DROP TABLE IF EXISTS `{db_name}`.`{table_name}`;CREATE TABLE IF NOT EXISTS `{db_name}`.`{table_name}`(  {table_column_infos})STORED AS PARQUETLOCATION '------/delta/{db_name}.db/{table_name}/data_date=${{data_date}}';INSERT OVERWRITE TABLE `{db_name}`.`{table_name}`SELECT {table_columns}FROM ( SELECT {table_columns}, _operation_type, row_number() OVER (PARTITION BY {row_keys} ORDER BY ts_ms DESC) as rank_num_        FROM (            SELECT {delta_columns}, operation_type as _operation_type, ts_ms            FROM delta_{db_name}_{table_name}            UNION ALL            SELECT {hive_columns}, 'c' as _operation_type, 0 as ts_ms            FROM `{db_name}`.`{table_name}_delta_history`      ) union_rank) ranked_dataWHERE rank_num_=1AND _operation_type <> 'd'