pyflink 统计词

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
from pyflink.common.typeinfo import Types

s_env = StreamExecutionEnvironment.get_execution_environment()
data = DataStream(s_env._j_stream_execution_environment.socketTextStream('192.168.137.200', 8899))
    #data.print()
def get_key():
        return '999'
class LogEvent:
    world=None

    def __init__(self,world,count):
        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.split('\s')
        word_counts = Counter(raw_message)
        print(word_counts)
        for word in word_counts: 
          word_counts = Counter(word)
        log_event = LogEvent(raw_message,word_counts)
        yield log_event.to_dict()

data.flat_map(MyMapFunction()).print()
s_env.execute('data')

Counter({'a a bb a': 1})
2> {'world': "['a a bb a']", 'count': "Counter({'a': 3, ' ': 3, 'b': 2})"}
Counter({'a b a c a d c f d': 1})
3> {'world': "['a b a c a d c f d']", 'count': "Counter({' ': 8, 'a': 3, 'c': 2, 'd': 2, 'b': 1, 'f': 1})"}