pyflink中Transform里的map的使用方法是什么？
写回答
class MyMapFunction(MapFunction):
def open(self, runtime_context: RuntimeContext):
# 数据量
self.count_state = runtime_context.get_state(ValueStateDescriptor("count_state", Types.INT()))
def map(self, value: Row):
self.count_state.update((self.count_state.value() or 0) + 1) # count_state += 1
return value
data_stream = data_stream.key_by(lambda x: x["id"]) \
.map(MyMapFunction(), output_type=Types.TUPLE([Types.STRING(), Types.STRING()]))
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
