实战：大数据Flink CDC同步Mysql数据到ElasticSearch

-- 假设我们有一个表 `order_info` 在 MySQL 数据库中，我们想要同步这个表的变更数据到 Elasticsearch。
 
-- 首先，我们需要创建一个源表，表示 MySQL 中的 `order_info` 表。
CREATE TABLE sourceTable (
  id INT,
  order_id STRING,
  order_time TIMESTAMP(3),
  user_id INT,
  product_id INT,
  amount DECIMAL(10, 2),
  status STRING
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'your_mysql_host_ip',
  'port' = '3306',
  'username' = 'your_username',
  'password' = 'your_password',
  'database-name' = 'your_database_name',
  'table-name' = 'order_info'
);
 
-- 然后，我们创建一个目标表，表示 Elasticsearch 中的索引。
CREATE TABLE sinkTable (
  id INT,
  order_id STRING,
  order_time TIMESTAMP(3),
  user_id INT,
  product_id INT,
  amount DECIMAL(10, 2),
  status STRING
) WITH (
  'connector' = 'elasticsearch-7',
  'hosts' = 'http://your_es_host_ip:9200',
  'index' = 'order_info_index',
  'sink.bulk-flush.max-actions' = '1', -- 为了示例，我们设置为1，表示每次处理一条数据。
  'sink.bulk-flush.max-size' = '1mb', -- 为了示例，我们设置为1mb。
  'sink.bulk-flush.interval' = '1s' -- 为了示例，我们设置为1秒。
);
 
-- 最后，我们执行同步操作。
INSERT INTO sinkTable
SELECT * FROM sourceTable;

这个示例代码展示了如何使用Flink SQL来同步MySQL中的数据变更日志到Elasticsearch。首先，我们定义了源表和目标表，然后通过INSERT INTO语句实现了数据的同步。这个例子简洁地展示了如何将数据从一个数据库同步到另一个搜索引擎，这是大数据处理中的一个常见需求。