[jira] [Created] (FLINK-24224) Table to stream, only the row datatype Stream works on CEP, and other POJOs, maps and jsonobjects datatype streams do not work, but any datatype stream to CEP can work by only stream api .

["YUJIANBO (Jira)" <ji...@apache.org>]

YUJIANBO created FLINK-24224:
--------------------------------

             Summary: Table to stream, only the row datatype Stream works on CEP, and other POJOs, maps and jsonobjects datatype streams do not work, but any datatype stream to CEP can work by only stream api .
                 Key: FLINK-24224
                 URL: https://issues.apache.org/jira/browse/FLINK-24224
             Project: Flink
          Issue Type: Bug
          Components: Library / CEP
    Affects Versions: 1.13.2, 1.12.0, 1.11.2
            Reporter: YUJIANBO


 

1、problem：*Table to stream*, only the *ROW* datatype Stream works on *CEP*, and other POJOs, maps and jsonobjects datatype streams do not work, but any datatype stream to CEP can work by only stream api .

2、version： I have tried 3 versions，such as 1.11.2、1.12.0、1.13.2

3、code：

（1）table to Stream  to  CEP   (only row datatype is ok,  other datatype Stream to CEP has no data print and it has no error message)
{code:java}
tableEnv.executeSql(creat_kafka_source);
tableEnv.executeSql(calculateSql);

Table tb = tableEnv.from("calculateSql");
String[] fieldNames = tb.getSchema().getFieldNames();
DataType[] fieldDataTypes = tb.getSchema().getFieldDataTypes();

KeyedStream<JSONObject, String> ds = tableEnv
        .toAppendStream(tb, Row.class)
        .map(new RichMapFunction<Row, JSONObject>() {
            Map<String, Object> map = new HashMap<>();

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                if (null == map) {
                    map = new HashMap<>();
                }
            }

            @Override
            public JSONObject map(Row value) throws Exception {
                //将数据key和value添加到map中
                RowParseUtil.setFieldAndValue2Map(map, fieldDataTypes, fieldNames, value);
                JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(map));
                map.clear();
                return jsonObject;
            }
        })
        .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<JSONObject>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(JSONObject element) {
                return element.getLongValue("wStart") * 1000;
            }
        }).keyBy(x -> x.getString("x_forwarded_for"));
//it has data to print
ds.print();

Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("begin")
        .where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                log.info("===================>" + value);
                return true;
            }
        }).timesOrMore(1).within(Time.seconds(10));

PatternStream<JSONObject> patternStream = CEP.pattern(ds, pattern);
//it has no data to print
patternStream.process(new PatternProcessFunction<JSONObject, String>() {
    @Override
    public void processMatch(Map<String, List<JSONObject>> match, Context ctx, Collector<String> out) throws Exception {
        out.collect("==========>>>>>>>" + match.toString());
    }
}).print();


{code}
(2) *Olny Stream API  to CEP* ( Any datatype ,  it is OK)
{code:java}
Properties proPs = kafkaUtil.getReceiveKfkProPs(receive_brokers, groupid);
FlinkKafkaConsumer<ConsumerRecord<String, String>> consumer =
        new FlinkKafkaConsumer<>(receive_topic, new KafkaRecordSchema(), proPs);
consumer.setStartFromEarliest();

SingleOutputStreamOperator<JSONObject> input = env.addSource(consumer)
        .map(x -> {
            return JSON.parseObject(x.value());
        })
        .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<JSONObject>(Time.milliseconds(10)) {
            @Override
            public long extractTimestamp(JSONObject element) {
                return element.getLongValue("ts");
            }
        })
        .keyBy(x -> x.getString("x_forwarded_for")+x.getString("request_uri"))
        .timeWindow(Time.seconds(1)).apply(new WindowFunction<JSONObject, JSONObject, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<JSONObject> input, Collector<JSONObject> out) throws Exception {
                Iterator<JSONObject> iterator = input.iterator();
                ArrayList<JSONObject> list = new ArrayList<>();
                int n = 0;
                while (iterator.hasNext()) {
                    n++;
                    JSONObject next = iterator.next();
                    list.add(next);
                }
                JSONObject jsonObject = list.get(0);
                jsonObject.put("ct",n);
                jsonObject.remove("ts");
                out.collect(jsonObject);
            }
        });

input.print();

//it is ok
Pattern<JSONObject, JSONObject> minInterval = Pattern
        .<JSONObject>begin("begin").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                return true;
            }
        }).timesOrMore(1).within(Time.seconds(10));

PatternStream<JSONObject> pattern = CEP.pattern(input, minInterval);
pattern.process(new PatternProcessFunction<JSONObject, String>() {
    @Override
    public void processMatch(Map<String, List<JSONObject>> map, Context context, Collector<String> out) throws Exception {
        out.collect("这个用户有嫌疑：====================>" + map.toString());
    }
}).print();
{code}
 



--
This message was sent by Atlassian Jira
(v8.3.4#803005)