Flink reads data from mysql and puts it into kafka to search for the full amount.

Continue with the previous article, import mysql data into kafka

Public static void main (String [] arg) throws Exception {TypeInformation [] fieldTypes = new TypeInformation [] {BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO}; String [] fieldNames = new String [] {"name", "address"}; RowTypeInfo rowTypeInfo = new RowTypeInfo (fieldTypes, fieldNames) JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat (). SetDrivername ("com.mysql.jdbc.Driver") .setDBUrl ("jdbc:mysql://ip:3306/tablespace?characterEncoding=utf8") .setUsername ("user"). SetPassword ("root") .setQuery ("select LOGIC_CODE, SHARE_LOG_CODE from table") .setRowTypeInfo (rowTypeInfo). Finish (); final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment () DataSource s = env.createInput (jdbcInputFormat); BatchTableEnvironment tableEnv = new BatchTableEnvironment (env, TableConfig.DEFAULT ()); tableEnv.registerDataSet ("T2", s); Table tapiResult = tableEnv.scan ("T2"); System.out.println ("schema is:"); tapiResult.printSchema (); Table query = tableEnv.sqlQuery ("select name, address from T2"); DataSet ds= tableEnv.toDataSet (query, Result.class) DataSet temp=ds.map (new MapFunction () {@ Override public String map (Result result) throws Exception {String name = result.name; String value = result.address; return name+ ":->:" + value;}}); logger.info ("read db end") KafkaOutputFormat kafkaOutput = KafkaOutputFormat.buildKafkaOutputFormat () .setBootstrapServers ("ip:9092"). SetTopic ("search_test_whk"). SetAcks ("all"). SetBatchSize ("1000") .setBufferMemory ("100000"). SetLingerMs ("1"). SetRetries ("2"). Finish (); temp.output (kafkaOutput); logger.info ("write kafka end"); env.execute ("Flink add data source") }