• -------------------------------------------------------------
  • ====================================

kafka之七 sinkTask

kafka dewbay 5年前 (2019-04-12) 2286次浏览 已收录 0个评论 扫描二维码

使用kafka connector 功能实现一个数据从kafka到 MySQL 的sinkTask

一:实现 JdbcSinkConnector 类

public class JdbcSinkConnector extends SinkConnector{
private String url;
private String driven;
private String userName;
private String passwd;
public void start(Map props) {
this.url = PropertyVerify.getOrElse(props, Constant_Global.URL, “jdbc:mysql://localhost/test”, “‘URL’ is null”);
this.driven = PropertyVerify.getOrElse(props, Constant_Global.DRIVEN, “com.mysql.jdbc.Driver”, “‘DRIVEN’ is null”);
this.userName = PropertyVerify.getOrElse(props, Constant_Global.USERNAME, “root”, “‘USERNAME’ is null”);
this.passwd = PropertyVerify.getOrElse(props, Constant_Global.PASSED, “root”, “‘PASSED’ is null”);
}
public Class taskClass() {
return JdbcSinkTask.class;
}
public List> taskConfigs(int maxTasks) {
ArrayList> configs = new ArrayList<>();
for(int i=0;i conf = new HashMap();
conf.put(“url”, url);
conf.put(“driven”, driven);
conf.put(“userName”, userName);
conf.put(“passwd”, passwd);
configs.add(conf);
}
return configs;
}
@Override
public String version() {
return AppInfoParser.getVersion();
}
@Override
public void stop() {
// TODO Auto-generated method stub
}
}
二:实现 JdbcSinkConnector 类
    
public class JdbcSinkTask extends SinkTask{
private static final Logger LOG = LoggerFactory.getLogger(JdbcSinkTask.class);
//private Connection conn = null;
public String shcema;
private JdbcDbWriter writer;
@Override
public String version() {
return new JdbcSinkConnector().version();
}
@Override
public void flush(Map map) {
LOG.info(“================flush Map start…………….===========================================================”);
}
@Override
public void put(Collection sinkRecords) {
if(sinkRecords.isEmpty()){
return;
}
try {
writer.write(sinkRecords,shcema,email);
} catch (SQLException | IOException e) {
try {
EmailUtil.init(Constant_Global.STMP, Constant_Global.EMAILUSER, Constant_Global.EMAILPASSWD, Constant_Global.EMAILTITAL, Constant_Global.EMAILADREE, email);
EmailUtil.send(” kafka sink 数据写入有问题 “);
} catch (MessagingException e1) {
e1.printStackTrace();
}
throw new JDBCConntorException(“数据写入有问题”);
}
}
@Override
public void start(Map pro) {
try {
DbPool.init(pro);
writer =new JdbcDbWriter();
} catch (PropertyVetoException e1) {
e1.printStackTrace();
LOG.info(“数据库配置异常=====”);
}
}
@Override
public void stop() {
}
}
三 :打包运行
      3.1 单机版运行,配置文件在kafka/config 目录下

          a: cp   connect-file-sink.properties    connect-jdbc-sink.properties

          b: vim connect-jdbc-sink.properties  配置如下

 # kafka connector properties
 name=canal-sink-connector  #定义 task 名称
 connector.class=com.trcloud.hamal.sink.jdbc.JdbcSinkConnector  #定义自己打包中的类
 tasks.max=1  #task 个数
 topics=words-out1   #消费的 topic
 url=jdbc:mysql://172.30.50.213/test   #数据库参数
 driven=com.mysql.jdbc.Driver #数据库驱动
 userName=root  #数据库用户名
 passwd=123456  #数据库密码

        c: 启动命令
         ./connect-standalone.sh   ../config/connect-standalone.properties  ../config/connect-jdbc-sink.properties
      3.2 单机版用于测试,生产环境建议使用分布式

            a: 配置文件 vim  jdbc-sink-distributed.properties         

bootstrap.servers=node1:6667,node2:6667,node3:6667
group.id=test-consumer-group
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.topic=connect-offsets
offset.flush.interval.ms=10000
config.storage.topic=configs-topic
status.storage.topic=connect-status
     3.2 启动命令 用 rest 接口启动
   

curl -X POST /connectors HTTP/1.1
Host: kafka.test.nd1
Content-Type: application/json
Accept: application/json
{
    “name”: “local-dw-sink”,
    “config”: {
    “connector.class”:”com.trcloud.hamal.sink.jdbc.JdbcSinkConnector”,
    “tasks.max”:”1″,
    “topics”:”sql-log” ,
    “url”:”jdbc:mysql://node4:3306/DW”,
    “driven”:”com.mysql.jdbc.Driver”,
    “userName”:”root”,
    “passwd”:”1234″
    }


露水湾 , 版权所有丨如未注明 , 均为原创丨本网站采用BY-NC-SA协议进行授权
转载请注明原文链接:kafka之七 sinkTask
喜欢 (2)
[]
分享 (0)
关于作者:
发表我的评论
取消评论

表情 贴图 加粗 删除线 居中 斜体 签到

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址