# 下载 Flink

wget https://archive.apache.org/dist/flink/flink-1.17.0/flink-1.17.0-bin-scala_2.12.tgz

# 解压

tar -xzf flink-1.17.0-bin-scala_2.12.tgz

# 移动到安装目录

mv flink-1.17.0 /ob/app/flink

# 编辑 /etc/profile

vim /etc/profile

# 添加以下内容

export FLINK_HOME=/ob/app/flink

export PATH=$PATH:$FLINK_HOME/bin

# 生效环境变量

source /etc/profile

obclient -h192.168.1.10 -P2881 -uroot@sys -p

ALTER SYSTEM SET enable_archive_log = ‘TRUE’;

SHOW PARAMETERS LIKE ‘enable_archive_log’;

— 输出：enable_archive_log | TRUE

// Flink CDC 配置代码

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.setParallelism(1);

DebeziumSourceFunction sourceFunction = MySQLSource.builder()

.hostname(“192.168.1.10”)

.port(2881)

.databaseList(“fgedudb”)

.tableList(“fgedudb.fgedu_order”)

.username(“fgedu”)

.password(“password”)

.deserializer(new JsonDebeziumDeserializationSchema())

.build();

DataStream stream = env.addSource(sourceFunction);

stream.addSink(KafkaSink.builder()

.setBootstrapServers(“192.168.1.11:9092”)

.setRecordSerializer(KafkaRecordSerializationSchema.builder()

.setTopic(“ob-order”)

.setValueSerializationSchema(new SimpleStringSchema())

.build())

.build());

env.execute(“OceanBase CDC to Kafka”);

// 实时计算代码

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.setParallelism(1);

DataStream stream = env.addSource(sourceFunction);

DataStream orderStream = stream.map(new MapFunction() {

@Override

public Order map(String value) throws Exception {

return JSON.parseObject(value, Order.class);

}

});

orderStream.keyBy(Order::getUserId)

.window(TumblingEventTimeWindows.of(Time.minutes(5)))

.sum(“amount”)

.print();

env.execute(“Real-time Order Analysis”);