【最佳实践】2个步骤教你从Mysql同步到Hive

-- 创建数据库
create database seatunnel;


-- 进入seatunnel数据库
use seatunnel;


-- 创建表
create table day_test(
dname varchar(64),
dage int
);


-- 插入数据
insert into day_test values('张三',20);
insert into day_test values('李四',18);
insert into day_test values('王二',29);
insert into day_test values('麻子',22);

#打开hive
hive


#新建数据库
create database mydemo;


#进入库
use mydemo;


#新建表
create table hive_mysql(
hname varchar(64),
hage int
);


#查看当前表的内容
select * from hive_mysql;

hive --service metastore &

#进入Seatunnel目录下的conf
cd seatunnel/apache-seatunnel-incubating-2.3.0/conf


#复制配置文件并改名，变为我们后面的启动文件
cp seatunnel/apache-seatunnel-incubating-2.3.0/conf/seatunnel.streaming.conf.template seatunnel/apache-seatunnel-incubating-2.3.0/conf/example01.conf


#打开文件修改
vi example01.conf


#保存并退出
:wq

env {
    execution.parallelism = 1
}
# 在source所属的块中配置数据源
source {
Jdbc {
        driver = "com.mysql.cj.jdbc.Driver"
        url = "jdbc:mysql://127.0.0.1:3306/seatunnel?serverTimezone=GMT%2b8&characterEncoding=utf-8"
        user = "root"
        password = "123456"
        query = "select * from day_test"
    }
}
# 在transform的块中声明转换插件
transform {


}
# 在sink块中声明要输出到哪
sink {
     Hive {
    table_name = "mydemo.hive_mysql"
    metastore_uri = "thrift://127.0.0.1:9083"
    schema {
      fields {
               hname = string
                hage= int
            }
          }
  }
}

cd seatunnel/apache-seatunnel-incubating-2.3.0


#用我们刚配置的文件去启动作业
./bin/start-seatunnel-flink-connector-v2.sh --config ./config/example01.conf

select * from hive_mysql;

sink {
    kafka {
      topic = "seatunnel"
      bootstrap.servers = "127.0.0.1:9092"
      partition = 3
      format = json
      kafka.request.timeout.ms = 60000
      semantics = EXACTLY_ONCE
  }
}

#kafka要是json格式
env {
    execution.parallelism = 1
}
# 在source所属的块中配置数据源
source {
    Kafka {
    result_table_name = "kafka_name"
    schema = {
      fields {
        id = "int"
        name = "string"
        age = "int"
      }
    }
    format = json
    field_delimiter = "#"
    topic = "mybate2"
    bootstrap.servers = "127.0.0.1:9092"
    kafka.max.poll.records = 500
    kafka.client.id = 127.0.0.1
  }
}
# 在transform的块中声明转换插件
transform {


}
# 在sink块中声明要输出到哪
sink {
         Hive {
    table_name = "mydemo.hive_mysql"
    metastore_uri = "thrift://127.0.0.1:9083"
    schema {
      fields {
                hid = int
               hname = string
                hage= int
            }
          }
  }
}

vim kafka_stop.sh

#!/bin/sh
#启动kafka挖取hive
./bin/start-seatunnel-flink-connector-v2.sh --config ./config/example01.conf


#等3秒后执行
sleep 3


#启动kafka同步数据到hive
./bin/start-seatunnel-flink-connector-v2.sh --config ./config/kafka_hive.conf

chmod +x cdc_hive.sh

sh cdc_hive.sh