Shulou(Shulou.com)06/03 Report--

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") }

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.