使用flink cdc，用 datastream API 能否实现相同的效果？

使用Flink CDC实现订单和商品表合并成宽表并写入Elasticsearch

1. 使用Flink SQL实现宽表合并与写入

在Flink SQL中，可以通过CDC（Change Data Capture）捕获订单表和商品表的变更数据，并通过JOIN操作将两张表合并为宽表，最终写入Elasticsearch。以下是实现的关键步骤：

• 创建CDC源表：分别定义订单表和商品表作为CDC源表，捕获其全量和增量数据。
• 定义宽表结构：通过SQL语句将订单表和商品表进行JOIN操作，生成宽表。
• 写入Elasticsearch：将宽表数据写入Elasticsearch目标表。

示例代码如下：

-- 创建订单表CDC源表
CREATE TABLE orders (
    order_id BIGINT,
    user_id STRING,
    product_id BIGINT,
    order_time TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '<your-mysql-host>',
    'port' = '3306',
    'username' = '<your-username>',
    'password' = '<your-password>',
    'database-name' = '<your-database>',
    'table-name' = 'orders'
);

-- 创建商品表CDC源表
CREATE TABLE products (
    product_id BIGINT,
    product_name STRING,
    category STRING,
    PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '<your-mysql-host>',
    'port' = '3306',
    'username' = '<your-username>',
    'password' = '<your-password>',
    'database-name' = '<your-database>',
    'table-name' = 'products'
);

-- 定义宽表并写入Elasticsearch
CREATE TABLE es_sink (
    order_id BIGINT,
    user_id STRING,
    product_id BIGINT,
    product_name STRING,
    category STRING,
    order_time TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'elasticsearch-7',
    'hosts' = '<your-es-host>',
    'index' = 'order_product_wide_table'
);

-- 合并订单表和商品表并写入Elasticsearch
INSERT INTO es_sink
SELECT 
    o.order_id,
    o.user_id,
    o.product_id,
    p.product_name,
    p.category,
    o.order_time
FROM orders AS o
LEFT JOIN products FOR SYSTEM_TIME AS OF o.proctime AS p
ON o.product_id = p.product_id;

此方法利用了Flink SQL的CDC功能和JOIN能力，能够高效地完成宽表合并和写入操作。

2. 使用Datastream API实现相同效果

在Datastream API中，也可以实现类似的功能，但需要手动处理CDC数据流、JOIN逻辑以及写入Elasticsearch的操作。以下是实现的关键步骤：

• 构建CDC Source：使用MySQL CDC Source捕获订单表和商品表的变更数据。
• 实现JOIN逻辑：通过KeyedCoProcessFunction或IntervalJoin等算子实现订单表和商品表的关联。
• 写入Elasticsearch：使用Elasticsearch Sink将合并后的宽表数据写入Elasticsearch。

示例代码如下：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.connector.elasticsearch.sink.ElasticsearchSink;
import org.apache.http.HttpHost;

public class OrderProductWideTable {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 构建订单表CDC Source
        MySqlSource<String> orderSource = MySqlSource.<String>builder()
            .hostname("<your-mysql-host>")
            .port(3306)
            .databaseList("<your-database>")
            .tableList("<your-database>.orders")
            .username("<your-username>")
            .password("<your-password>")
            .deserializer(new JsonDebeziumDeserializationSchema())
            .build();

        // 构建商品表CDC Source
        MySqlSource<String> productSource = MySqlSource.<String>builder()
            .hostname("<your-mysql-host>")
            .port(3306)
            .databaseList("<your-database>")
            .tableList("<your-database>.products")
            .username("<your-username>")
            .password("<your-password>")
            .deserializer(new JsonDebeziumDeserializationSchema())
            .build();

        // 处理订单表和商品表的JOIN逻辑
        env.addSource(orderSource)
            .keyBy(order -> order.getField("product_id"))
            .connect(env.addSource(productSource).keyBy(product -> product.getField("product_id")))
            .process(new KeyedCoProcessFunction<>() {
                // 实现JOIN逻辑
            })
            .addSink(new ElasticsearchSink.Builder<>(
                Collections.singletonList(new HttpHost("<your-es-host>", 9200, "http")),
                new ElasticsearchSinkFunction<>() {
                    // 实现写入Elasticsearch逻辑
                }
            ).build());

        env.execute("Order and Product Wide Table");
    }
}

在此实现中，您需要手动处理CDC数据流的解析、JOIN逻辑的实现以及Elasticsearch的写入操作。

3. 两种方式的对比

4. 重要提醒

• CDC配置限制：确保MySQL的Binlog已正确配置，避免因过滤器导致部分表无法同步。
• 依赖管理：如果使用Datastream API，需自行管理Flink CDC连接器及其依赖，建议使用稳定版本的JAR包。
• Elasticsearch版本兼容性：确保使用的Elasticsearch连接器与目标Elasticsearch版本兼容。

通过上述两种方式，您可以根据实际需求选择合适的方法实现订单和商品表的宽表合并与写入Elasticsearch。