用python调用PyFlink 1.17.1版本RichSinkFunction代码，用python调用PyFlink 1.17.1版本SinkFunction代码

以下是使用 Python 调用 PyFlink 1.17.1 版本的 RichSinkFunction 和 SinkFunction 的示例代码：

RichSinkFunction 代码示例：

from pyflink.common.serialization import SimpleStringEncoder
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import StreamingFileSink
from pyflink.datastream.functions import RichSinkFunction
from pyflink.datastream.stream import DataStream


class MySink(RichSinkFunction):

    def __init__(self, file_path):
        self.file_path = file_path

    def open(self, runtime_context):
        self.writer = open(self.file_path, 'w')

    def invoke(self, value, context):
        # 将数据写入外部存储系统
        self.writer.write(str(value) + '\n')

    def close(self):
        self.writer.close()


env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)

# 创建数据流
data_stream = env.from_collection([(1, 'hello'), (2, 'world'), (3, 'flink')], Types.TUPLE([Types.INT(), Types.STRING()]))

# 定义输出路径
output_path = '/path/to/output'

# 创建 StreamingFileSink
sink = StreamingFileSink \
    .for_row_format(output_path, SimpleStringEncoder()) \
    .with_bucket_assigner(None) \
    .with_bucket_check_interval(1000) \
    .with_part_prefix('prefix') \
    .with_part_suffix('.txt') \
    .build()

# 添加 sink
data_stream.add_sink(MySink(output_path)).set_parallelism(1)

# 执行任务
env.execute('PyFlink RichSinkFunction Example')

SinkFunction 代码示例：

from pyflink.common.serialization import SimpleStringEncoder
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import StreamingFileSink
from pyflink.datastream.functions import SinkFunction
from pyflink.datastream.stream import DataStream


class MySink(SinkFunction):

    def __init__(self, file_path):
        self.file_path = file_path
        self.writer = None

    def open(self, runtime_context):
        self.writer = open(self.file_path, 'w')

    def invoke(self, value, context):
        # 将数据写入外部存储系统
        self.writer.write(str(value) + '\n')

    def close(self):
        self.writer.close()


env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)

# 创建数据流
data_stream = env.from_collection([(1, 'hello'), (2, 'world'), (3, 'flink')], Types.TUPLE([Types.INT(), Types.STRING()]))

# 定义输出路径
output_path = '/path/to/output'

# 创建 StreamingFileSink
sink = StreamingFileSink \
    .for_row_format(output_path, SimpleStringEncoder()) \
    .with_bucket_assigner(None) \
    .with_bucket_check_interval(1000) \
    .with_part_prefix('prefix') \
    .with_part_suffix('.txt') \
    .build()

# 添加 sink
data_stream.add_sink(MySink(output_path)).set_parallelism(1)

# 执行任务
env.execute('PyFlink SinkFunction Example')

在这两个示例中，我们都定义了一个自定义的 sink 类（MySink），并在其中实现了 RichSinkFunction 或 SinkFunction 接口的方法。同时，我们还使用了 StreamingFileSink 将数据写入外部存储系统。这里我们使用了 for_row_format 方法定义了输出路径、编码方式等参数，然后调用 build 方法构建 StreamingFileSink 对象，最后通过 add_sink 方法将自定义的 sink 对象添加到数据流中。最后，我们调用 execute 方法执行任务。