Blink流式计算-Kafka接入demo

//定义解析Kakfa message的UDTF

CREATE FUNCTION myParse AS 'com.xxxxxx.MyKafkaUDTF';

CREATE FUNCTION myUdf AS 'com.xxxxxxx.MyWaterMarkUDTF';

//注意：kafka源表DDL字段必须与以下例子一致

create table my_input (

messageKey VARBINARY,

message VARBINARY,

topic varchar,

partition int,

offset bigint,

ctTime AS TO_TIMESTAMP (myUdf (message)),

//注意计算里的类型必须为timestamp才能在做watermark。

WATERMARK wk FOR ctTime AS WITHOFFSET (ctTime, 2000) --为rowtime定义watermark

) WITH (

type = 'KAFKA08',

topic = 'myTopic',

group.id = 'mGroup',

extraConfig = 'bootstrap.servers=127.0.0.1:9092',

zookeeper.connect = '127.0.0.1:2181',

startupMode = 'EARLISET'

);

-- 滚动窗口 group by prodId

CREATE VIEW input_view01 (

windowStart,

windowEnd,

prodId,

prodName,

prodNumber

) AS

SELECT

HOP_START (S.ctTime, INTERVAL '30' SECOND, INTERVAL '2' MINUTE),

HOP_END (S.ctTime, INTERVAL '30' SECOND, INTERVAL '2' MINUTE),

T.prodId as prodId,

T.prodName as prodName,

count (*) as prodNumber

from

my_input as S,

LATERAL TABLE (myParse (message)) as T (

id,

prodId,

prodName,

createdAt,

updatedAt

)

Group BY HOP (S.ctTime, INTERVAL '30' SECOND, INTERVAL '2' MINUTE), T.prodId, T.prodName;

CREATE VIEW input_view60 (

id,

prodId,

prodName,

createdAt,

updatedAt

) AS

SELECT

T.id,

T.prodId,

T.prodName,

T.createdAt,

T.updatedAt

from

my_input as S,

LATERAL TABLE (myParse (message)) as T (

id,

goCs,

prodId,

prodName,

createdAt,

updatedAt

);

-- 结果print

create table outprint01(

prodId bigint,

prodName varchar,

prodNumber bigint

)with(

type = 'print'

);

insert into outprint01

select prodId , prodName , prodNumber

from input_view01;

-- 结算结果写入Kafka

create table result_kafka (

messageKey VARBINARY,

message VARBINARY,

PRIMARY KEY (messageKey)

) with (

type = 'KAFKA08',

topic = 'myResultTopic',

extraConfig='bootstrap.servers=127.0.0.1:9092',

zookeeper.connect = '127.0.0.1:2181',

startupMode='EARLISET'

);

//此处的结果输出，可以考虑将结果组装成字符串，中间用|隔开，接收方再解析

INSERT INTO

result_kafka

SELECT

cast(prodId as VARBINARY) as messageKey,

cast(prodName as VARBINARY) as message

FROM

input_view01;

MyKafkaUDTF写法：

package com.xxxxxxxx;

import com.alibaba.fastjson.JSONObject;

import org.apache.flink.table.functions.TableFunction;

import org.apache.flink.table.types.DataType;

import org.apache.flink.table.types.DataTypes;

import org.apache.flink.types.Row;

import java.io.UnsupportedEncodingException;

import java.sql.Timestamp;

public class MyKafkaUDTF extends TableFunction {

public void eval(byte[] message) {

try {

String msg = new String(message, "UTF-8");

System.out.println("收到的消息:"+msg);

try {

JSONObject jsonObject = JSONObject.parseObject(msg);

if (jsonObject != null) {

//id

Long id = jsonObject.getLong("id");

//prodId

Long prodId = jsonObject.getLong("prodId");

//prodName

String prodName = jsonObject.getString("prodName ");

Long createAt = jsonObject.getLong("createdAt");

Long updatedAt = jsonObject.getLong("updatedAt");

//创建时间时间戳

Timestamp createAtTimeStamp = new Timestamp(createAt);

Timestamp updatedAtTimeStamp = new Timestamp(updatedAt);

Row row = new Row(8);

row.setField(0, id);

row.setField(1, prodId);

row.setField(2, prodName);

row.setField(3, createAtTimeStamp );

row.setField(4, updatedAtTimeStamp );

System.out.println("message str ==>" + row.toString());

collect(row);

}

} catch (Exception e) {

e.printStackTrace();

System.out.println(" error. Input data " + msg + "is not json string");

}

} catch (UnsupportedEncodingException e) {

e.printStackTrace();

}

}

@Override

// 如果返回值是Row，就必须重载实现这个方法，显式地告诉系统返回的字段类型

public DataType getResultType(Object[] arguments, Class[] argTypes) {

return DataTypes.createRowType(

DataTypes.LONG,

DataTypes.LONG,

DataTypes.STRING,

DataTypes.TIMESTAMP,

DataTypes.TIMESTAMP);

}

}

package xxxxxxx;

import com.alibaba.fastjson.JSONObject;

import org.apache.flink.table.functions.ScalarFunction;

import java.text.SimpleDateFormat;

import java.util.Date;

public class MyWaterMarkUDTF extends ScalarFunction {

public String eval(byte[] message) {

try {

String msg = new String(message, "UTF-8");

JSONObject data = JSONObject.parseObject(msg);

System.out.println("time:"+data.getString("createdAt"));

Long createAtLong = data.getLong("createdAt");

SimpleDateFormat parser = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

String createTimeStr = parser.format(new Date(createAtLong));

return createTimeStr;

} catch (Exception e) {

e.printStackTrace();

}

return null;

}

//可选，close方法可以不写

@Override

public void close() {

}

}