批量读取_Flink1.11版本读取kafka数据批量写入mysql

主要代码

package cn.nanxiuzi.kafka.kafka2mysql;import cn.nanxiuzi.kafka.KafkaDic;import com.google.common.collect.Lists;import org.apache.flink.api.common.functions.MapFunction;import org.apache.flink.api.common.serialization.SimpleStringSchema;import org.apache.flink.streaming.api.CheckpointingMode;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;import org.apache.flink.streaming.api.windowing.time.Time;import org.apache.flink.streaming.api.windowing.windows.TimeWindow;import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;import org.apache.flink.util.Collector;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.util.ArrayList;import java.util.List;import java.util.Properties;/** * 读取kafka之后写入mysql */public class Kafka2Mysql { public static final Logger logger &＃61; LoggerFactory.getLogger(Kafka2Mysql.class); public static void main(String[] args) throws Exception { StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment(); //checkpoint的设置 //每隔10s进行启动一个检查点【设置checkpoint的周期】 env.enableCheckpointing(10000); //设置模式为&＃xff1a;exactly_one&＃xff0c;仅一次语义 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); //确保检查点之间有1s的时间间隔【checkpoint最小间隔】 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000); //检查点必须在10s之内完成&＃xff0c;或者被丢弃【checkpoint超时时间】 env.getCheckpointConfig().setCheckpointTimeout(10000); //同一时间只允许进行一次检查点 env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); //表示一旦Flink程序被cancel后&＃xff0c;会保留checkpoint数据&＃xff0c;以便根据实际需要恢复到指定的checkpoint //env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); //设置statebackend,将检查点保存在hdfs上面&＃xff0c;默认保存在内存中。这里先保存到本地// env.setStateBackend(new FsStateBackend("file:///Users/temp/cp/")); Properties ppt &＃61; new Properties(); ppt.setProperty("bootstrap.servers", KafkaDic.Kafka_ADDRESS_COLLECTION); ppt.setProperty("group.id", KafkaDic.CONSUMER_GROUP_ID); ppt.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); ppt.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); FlinkKafkaConsumer flinkKafkaConsumer &＃61; new FlinkKafkaConsumer<>(KafkaDic.CONSUMER_TOPIC, new SimpleStringSchema(), ppt); flinkKafkaConsumer.setStartFromLatest(); DataStreamSource kafkaSource &＃61; env.addSource(flinkKafkaConsumer).setParallelism(1); kafkaSource.map(new MapFunction() { &＃64;Override public String map(String o) throws Exception { return o.split(",")[0]; } }).timeWindowAll(Time.seconds(10)).process(new ProcessAllWindowFunction, TimeWindow>() { &＃64;Override public void process(Context context, Iterable elements, Collector> out) throws Exception { ArrayList strings &＃61; Lists.newArrayList(elements); if (strings.size() > 0) { out.collect(strings); } } }).addSink(new MysqlSink()).setParallelism(1); env.execute("MysqlSink"); }}

Kafka生产者代码

package cn.nanxiuzi.kafka;import org.apache.kafka.clients.producer.KafkaProducer;import org.apache.kafka.clients.producer.ProducerRecord;import java.util.Properties;public class MyKafkaProducer { public static void main(String[] args) { Properties props &＃61; new Properties(); props.put("bootstrap.servers", KafkaDic.Kafka_ADDRESS_COLLECTION); props.put("acks", "all"); props.put("retries", 0); props.put("batch.size", 16384); props.put("linger.ms", 1); props.put("buffer.memory", 33554432); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); KafkaProducer kafkaProducer &＃61; new KafkaProducer<>(props); for(int i&＃61;0;i<10000;i&＃43;&＃43;){ String messageContext&＃61;String.format("姓名%s,广东深圳%s,身高%s,体重%s,电话%s",Integer.toString(i),Integer.toString(i),Integer.toString(i),Integer.toString(i),Integer.toString(i)); kafkaProducer.send(new ProducerRecord(KafkaDic.PRODUCER_TOPIC,Integer.toString(i),messageContext)); System.out.println("sented:"&＃43;messageContext); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } kafkaProducer.close(); }}

DBConnectUtil

package cn.nanxiuzi.kafka.kafka2mysql;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.sql.DriverManager;import java.sql.SQLException;import java.sql.Connection;/** * Created with IntelliJ IDEA. * User: zzy * Date: 2019/5/28 * Time: 8:58 PM * To change this template use File | Settings | File Templates. */public class DBConnectUtil { private static final Logger log &＃61; LoggerFactory.getLogger(DBConnectUtil.class); /** * 获取连接 * * &＃64;param url * &＃64;param user * &＃64;param password * &＃64;return * &＃64;throws SQLException */ public static Connection getConnection(String url, String user, String password) throws SQLException { Connection conn &＃61; null; try { Class.forName("com.mysql.jdbc.Driver"); } catch (ClassNotFoundException e) { log.error("获取mysql.jdbc.Driver失败"); e.printStackTrace(); } try { conn &＃61; DriverManager.getConnection(url, user, password); log.info("获取连接:{} 成功...",conn); }catch (Exception e){ log.error("获取连接失败&＃xff0c;url:" &＃43; url &＃43; ",user:" &＃43; user); } //设置手动提交 //conn.setAutoCommit(false); return conn; } /** * 提交事物 */ public static void commit(Connection conn) { if (conn !&＃61; null) { try { conn.commit(); } catch (SQLException e) { log.error("提交事物失败,Connection:" &＃43; conn); e.printStackTrace(); } finally { close(conn); } } } /** * 事物回滚 * * &＃64;param conn */ public static void rollback(Connection conn) { if (conn !&＃61; null) { try { conn.rollback(); } catch (SQLException e) { log.error("事物回滚失败,Connection:" &＃43; conn); e.printStackTrace(); } finally { close(conn); } } } /** * 关闭连接 * * &＃64;param conn */ public static void close(Connection conn) { if (conn !&＃61; null) { try { conn.close(); } catch (SQLException e) { log.error("关闭连接失败,Connection:" &＃43; conn); e.printStackTrace(); } } }}

输出

pom.xml

4.0.0modelVersion> cn.nanxiuzigroupId> myflinkartifactId> 1.0-SNAPSHOTversion> 1.11.2flink-version> 1.2.17log4j-version> properties> org.apache.flinkgroupId> flink-javaartifactId> ${flink-version}version> providedscope> dependency> org.apache.flinkgroupId> flink-coreartifactId> ${flink-version}version> providedscope> dependency> org.apache.flinkgroupId> flink-streaming-java_2.11artifactId> ${flink-version}version> providedscope> dependency> org.apache.flinkgroupId> flink-connector-kafka_2.11artifactId> ${flink-version}version> dependency> org.apache.flinkgroupId> flink-hbase_2.11artifactId> 1.10.2version> dependency> org.apache.flinkgroupId> flink-connector-filesystem_2.11artifactId> ${flink-version}version> dependency> org.apache.flinkgroupId> flink-connector-jdbc_2.11artifactId> ${flink-version}version> dependency> org.apache.flinkgroupId> flink-clients_2.11artifactId> ${flink-version}version> dependency> com.github.javafakergroupId> javafakerartifactId> 1.0.2version> dependency> log4jgroupId> log4jartifactId> ${log4j-version}version> dependency> com.fasterxml.jackson.coregroupId> jackson-databindartifactId> 2.11.2version> dependency> com.alibabagroupId> fastjsonartifactId> 1.2.73version> dependency> mysqlgroupId> mysql-connector-javaartifactId> 5.1.25version> dependency> dependencies> org.apache.maven.pluginsgroupId> maven-shade-pluginartifactId> 3.1.1version> packagephase> shadegoal> goals> com.google.code.findbugs:jsr305exclude> org.slf4j:*exclude> log4j:*exclude> excludes> artifactSet> *:*artifact> META-INF/*.SFexclude> META-INF/*.DSAexclude> META-INF/*.RSAexclude> excludes> filter> filters> my.programs.main.clazzmainClass> transformer> transformers> configuration> execution> executions> plugin> org.apache.maven.pluginsgroupId> maven-compiler-pluginartifactId> 8source> 8target> configuration> plugin> plugins> build>project>