Flink+kafka 流數(shù)據(jù) 使用FlinkSql 計(jì)算 10秒滾動(dòng)窗口內(nèi) 用戶(hù)點(diǎn)擊次數(shù)，之后自定義 sink To mysql。

Flink版本為1.6.1?

代碼如下：

FlinkSqlWindowUserPv.java

import java.sql.Timestamp;

import java.util.Properties;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.common.serialization.SimpleStringSchema;

import org.apache.flink.api.common.typeinfo.Types;

import org.apache.flink.api.java.tuple.Tuple3;

import org.apache.flink.api.java.tuple.Tuple5;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;

import org.apache.flink.table.api.Table;

import org.apache.flink.table.api.TableConfig;

import org.apache.flink.table.api.java.StreamTableEnvironment;

import pojo.UserPvEntity;

public class FlinkSqlWindowUserPv{

? ? public static void main(String[] args) throws Exception {

? ? StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

? ? env.setParallelism(8);

? ? TableConfig tc = new TableConfig();

? ? StreamTableEnvironment tableEnv = new StreamTableEnvironment(env, tc);

? ? ? ? Properties properties = new Properties();

? ? ? ? properties.put("bootstrap.servers", "127.0.0.1:9092");

? ? ? ? properties.put("zookeeper.connect", "127.0.0.1:2181");

? ? ? ? properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

? ? ? ? properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

? ? ? ? properties.put("group.id", "test6");

? ? ? ? FlinkKafkaConsumer010<String> myConsumer = new FlinkKafkaConsumer010<String>("myItems_topic5", new SimpleStringSchema(),

? ? ? ? ? ? ? ? properties);

? ? ? ? DataStream<String> stream = env.addSource(myConsumer);

? ? ? ? DataStream<Tuple5<String, String, String, String, Long>> map = stream.map(new MapFunction<String, Tuple5<String, String, String, String,Long>>() {

? ? ? ? private static final long serialVersionUID = 1471936326697828381L;

@Override

public Tuple5<String, String, String, String,Long> map(String value) throws Exception {

String[] split = value.split(" ");

return new Tuple5<String, String, String, String,Long>(split[0],split[1],split[2],split[3],Long.valueOf(split[4])*1000);

}

});

? ? ? ? map.print(); //打印流數(shù)據(jù)

? ? ? ? //注冊(cè)為user表

? ? ? ? tableEnv.registerDataStream("Users", map, "userId,itemId,categoryId,behavior,timestampin,proctime.proctime");

? ? ? ? //執(zhí)行sql查詢(xún)? ? 滾動(dòng)窗口 10秒? ? 計(jì)算10秒窗口內(nèi)用戶(hù)點(diǎn)擊次數(shù)

? ? ? ? Table sqlQuery = tableEnv.sqlQuery("SELECT TUMBLE_END(proctime, INTERVAL '10' SECOND) as processtime,"

? ? ? ? + "userId,count(*) as pvcount "

? ? ? ? + "FROM Users "

? ? ? ? + "GROUP BY TUMBLE(proctime, INTERVAL '10' SECOND), userId");

? ? ? ? //Table 轉(zhuǎn)化為 DataStream

? ? ? ? DataStream<Tuple3<Timestamp, String, Long>> appendStream = tableEnv.toAppendStream(sqlQuery,Types.TUPLE(Types.SQL_TIMESTAMP,Types.STRING,Types.LONG));

? ? ? ? appendStream.print();

? ? ? ? //sink to mysql

? ? ? ? appendStream.map(new MapFunction<Tuple3<Timestamp,String,Long>, UserPvEntity>() {

private static final long serialVersionUID = -4770965496944515917L;

@Override

public UserPvEntity map(Tuple3<Timestamp, String, Long> value) throws Exception {

return new UserPvEntity(Long.valueOf(value.f0.toString()),value.f1,value.f2);

}

}).addSink(new SinkUserPvToMySQL2());

? ? ? ? env.execute("userPv from Kafka");

? ? }

}

?SinkUserPvToMySQL2.java

import java.sql.Connection;

import java.sql.DriverManager;

import java.sql.PreparedStatement;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import pojo.UserPvEntity;

public class SinkUserPvToMySQL2 extends RichSinkFunction<UserPvEntity> {

private static final long serialVersionUID = -4443175430371919407L;

PreparedStatement ps;

? ? private Connection connection;

? ? /**

? ? * open() 方法中建立連接，這樣不用每次 invoke 的時(shí)候都要建立連接和釋放連接

? ? *

? ? * @param parameters

? ? * @throws Exception

? ? */

? ? @Override

? ? public void open(Configuration parameters) throws Exception {

? ? ? ? super.open(parameters);

? ? ? ? connection = getConnection();

? ? ? ? String sql = "replace into t_user_pv(pvtime,userId, pvcount) values(?, ?, ?);";

? ? ? ? ps = this.connection.prepareStatement(sql);

? ? }

? ? @Override

? ? public void close() throws Exception {

? ? ? ? super.close();

? ? ? ? //關(guān)閉連接和釋放資源

? ? ? ? if (connection != null) {

? ? ? ? ? ? connection.close();

? ? ? ? }

? ? ? ? if (ps != null) {

? ? ? ? ? ? ps.close();

? ? ? ? }

? ? }

? ? /**

? ? * 每條數(shù)據(jù)的插入都要調(diào)用一次 invoke() 方法

? ? *

? ? * @param value

? ? * @param context

? ? * @throws Exception

? ? */

? ? @Override

? ? public void invoke(UserPvEntity userPvEntity, Context context) throws Exception {

? ? ? ? //組裝數(shù)據(jù)，執(zhí)行插入操作

? ? ps.setLong(1, userPvEntity.getTime());

? ? ps.setString(2, userPvEntity.getUserId());

? ? ? ? ps.setLong(3, userPvEntity.getPvcount());

? ? ? ? ps.executeUpdate();

? ? }

? ? private static Connection getConnection() {

? ? ? ? Connection con = null;

? ? ? ? try {

? ? ? ? ? ? Class.forName("com.mysql.jdbc.Driver");

? ? ? ? ? ? con = DriverManager.getConnection("jdbc:mysql://localhost:3306/myTable??useUnicode=true&characterEncoding=UTF-8","root","123456");

? ? ? ? } catch (Exception e) {

? ? ? ? ? ? System.out.println("-----------mysql get connection has exception , msg = "+ e.getMessage());

? ? ? ? }

? ? ? ? return con;

? ? }

}

結(jié)果展示：

---------------------

作者：麥香雞翅

來(lái)源：CSDN

原文：https://blog.csdn.net/qq_20672231/article/details/84936716

版權(quán)聲明：本文為博主原創(chuàng)文章，轉(zhuǎn)載請(qǐng)附上博文鏈接！