PyFlink -如何使用PyFlink推送数据到mongodb和redis？

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment
from pyflink.table.descriptors import Schema, Kafka, Json, Elasticsearch, FileSystem
from pyflink.table.types import DataTypes
from pyflink.table.udf import ScalarFunction
from pyflink.common.serialization import Encoder
from pyflink.common.typeinfo import Types
from pyflink.datastream.connectors import FlinkKafkaProducer
from pyflink.datastream import TimeCharacteristic
from pyflink.datastream.connectors import FlinkKafkaConsumer
import pymongo
import redis

env = StreamExecutionEnvironment.get_execution_environment()
env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
t_env = StreamTableEnvironment.create(env)

source_topic = "source_topic"
sink_topic = "sink_topic"
kafka_properties = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "test-group"
}

source_ddl = f"""
    CREATE TABLE source_table (
        ...
    ) WITH (
        'connector' = 'kafka',
        'topic' = '{source_topic}',
        'properties.bootstrap.servers' = '{kafka_properties["bootstrap.servers"]}',
        'properties.group.id' = '{kafka_properties["group.id"]}',
        'format' = 'json'
    )
"""
t_env.execute_sql(source_ddl)

source_table = t_env.from_path("source_table")

result_table = source_table...

mongo_client = pymongo.MongoClient("mongodb://localhost:27017/")
mongo_db = mongo_client["database_name"]
mongo_collection = mongo_db["collection_name"]

def mongodb_sink(data):
    mongo_collection.insert_one(data)

result_table.select("...").insert_into("mongodb_sink")

redis_client = redis.Redis(host="localhost", port=6379, db=0)

def redis_sink(data):
    redis_client.set("key", data)

result_table.select("...").insert_into("redis_sink")

env.execute("Job Name")