使用Flink CDC将Mysql中的数据实时同步到ES-Toy模板网

这篇具有很好参考价值的文章主要介绍了使用Flink CDC将Mysql中的数据实时同步到ES。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

前言

最近公司要搞搜索，需要把mysql中的数据同步到es中来进行搜索，由于公司已经搭建了flink集群，就打算用flink来做这个同步。本来以为很简单，跟着官网文档走就好了，结果没想到折腾了将近一周的时间……

我也是没想到，这玩意网上资源竟然这么少，找到的全部都是通过flink sql-client实现的，但这有个问题，当fink集群重启，JOB就没有了，没有办法通过savePointing来恢复。所以还是记录下。

代码

直接上代码：文章来源地址https://www.toymoban.com/news/detail-516097.html

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

        env.enableCheckpointing(3000);
	    env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink/savepointings");

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("CREATE TABLE orders (\n" +
                "   order_id INT,\n" +
                "   order_date TIMESTAMP(0),\n" +
                "   customer_name STRING,\n" +
                "   price DECIMAL(10, 5),\n" +
                "   product_id INT,\n" +
                "   order_status BOOLEAN,\n" +
                "   PRIMARY KEY (order_id) NOT ENFORCED\n" +
                " ) WITH (\n" +
                "   'connector' = 'mysql-cdc',\n" +
                "   'hostname' = 'localhost',\n" +
                "   'port' = '3306',\n" +
                "   'username' = 'root',\n" +
                "   'password' = '123456',\n" +
                "   'database-name' = 'mydb',\n" +
                "   'table-name' = 'orders'\n" +
                " );").await();

        tableEnv.executeSql("CREATE TABLE products (\n" +
                "    id INT,\n" +
                "    name STRING,\n" +
                "    description STRING,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = 'localhost',\n" +
                "    'port' = '3306',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'database-name' = 'mydb',\n" +
                "    'table-name' = 'products'\n" +
                "  );").await();

        tableEnv.executeSql("CREATE TABLE enriched_orders (\n" +
                "   order_id INT,\n" +
                "   order_date TIMESTAMP(0),\n" +
                "   customer_name STRING,\n" +
                "   price DECIMAL(10, 5),\n" +
                "   product_id INT,\n" +
                "   order_status BOOLEAN,\n" +
                "   product_name STRING,\n" +
                "   product_description STRING,\n" +
                "   PRIMARY KEY (order_id) NOT ENFORCED\n" +
                " ) WITH (\n" +
                "     'connector' = 'elasticsearch-7',\n" +
                "     'hosts' = 'http://localhost:9200',\n" +
                "     'index' = 'enriched_orders_lhc'\n" +
                " );");

        tableEnv.executeSql("INSERT INTO enriched_orders\n" +
                " SELECT o.*, p.name, p.description\n" +
                " FROM orders AS o\n" +
                " LEFT JOIN products AS p ON o.product_id = p.id");

        env.execute("Mysql to ES");
    }