package com.atguigu.flink.streamx;
import com.streamxhub.streamx.flink.core.StreamEnvConfig;
import com.streamxhub.streamx.flink.core.java.function.SQLFromFunction;
import com.streamxhub.streamx.flink.core.java.sink.JdbcSink;
import com.streamxhub.streamx.flink.core.java.source.KafkaSource;
import com.streamxhub.streamx.flink.core.scala.StreamingContext;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
/**
* TODO
*
* @author cjp
* @version 1.0
*/
public class StreamxMySQLSinkDemo {
public static void main(String[] args) {
StreamEnvConfig envConfig = new StreamEnvConfig(args, null);
StreamingContext context = new StreamingContext(envConfig);
SingleOutputStreamOperator<WaterSensor> kafkaDS = new KafkaSource<String>(context)
.alias("kafka1")
.topic("s1")
.getDataStream()
.map(record -> {
String[] split = record.value().split(",");
return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[1]));
});
new JdbcSink<WaterSensor>(context)
.sql(new SQLFromFunction<WaterSensor>() {
@Override
public String from(WaterSensor waterSensor) {
return String.format(
"insert into sensor(id,ts,vc) values('%s',%d,%d)",
waterSensor.getId(), waterSensor.getTs(), waterSensor.getVc());
}
})
.sink(kafkaDS);
context.start();
}
}
streamx-demo-master.zip （19个子文件） 
streamx-demo-master 
pom.xml 8KB assembly bin shutdown.sh 2KB startup.sh 2KB setclasspath.sh 4KB streamx.sh 8KB conf application.yml 3KB sql.yml 481B logs 2022-04-14 warn.0.log 0B trace.0.log 0B info.0.log 0B error.0.log 0B debug.0.log 0B assembly.xml 1KB src main resources logback.xml 6KB java com atguigu flink streamx WaterSensor.java 904B StreamxKafkaDemo.java 1KB StreamxKafkaDemo2.java 985B StreamxMySQLSinkDemo.java 2KB StreamXSqlDemo.java 462B
评论0