import json
import re
import logging
import sys
from collections import Counter
from pyflink.datastream import DataStream, StreamExecutionEnvironment
from pyflink.datastream.functions import RuntimeContext, FlatMapFunction, MapFunction,FilterFunction
from pyflink.common.typeinfo import Types
s_env = StreamExecutionEnvironment.get_execution_environment()
data = DataStream(s_env._j_stream_execution_environment.socketTextStream('192.168.137.201', 8899))
# data.print()
def get_key():
return '999'
class LogEvent:
world = None
def __init__(self, world):
self.world = world
# self.count = count
def to_dict(self):
return {
"world": str(self.world)
# "count": str(self.count)
}
class MyMapFunction(FlatMapFunction):
def open(self, runtime_context: RuntimeContext):
pass
def flat_map(self, raw_message):
raw_message = raw_message.upper()
dict1 = {}
dict1['value'] = raw_message +'!!!'
yield dict1
class myFilterFunction(FilterFunction):
def filter(self,raw_message):
if re.findall('AAA',raw_message):
return raw_message
data.filter(myFilterFunction()).print()
s_env.execute('data')