# 编写Flink作业
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import com.alibaba.fastjson.JSONObject;

public class TiDBChangeProcessor {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        KafkaSource<String> source = KafkaSource.<String>builder()
            .setBootstrapServers("192.168.1.15:9092")
            .setTopics("tidb-binlog")
            .setGroupId("flink-group")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();
        
        DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
        
        stream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                JSONObject json = JSONObject.parseObject(value);
                String database = json.getString("database");
                String table = json.getString("table");
                String type = json.getString("type");
                
                if (database.equals("fgedudb") && table.equals("fgedu_users")) {
                    if (type.equals("INSERT")) {
                        JSONObject data = json.getJSONArray("data").getJSONObject(0);
                        System.out.println("New user added: " + data.getString("username"));
                    } else if (type.equals("UPDATE")) {
                        JSONObject data = json.getJSONArray("data").getJSONObject(0);
                        System.out.println("User updated: " + data.getString("username"));
                    } else if (type.equals("DELETE")) {
                        JSONObject old = json.getJSONArray("old").getJSONObject(0);
                        System.out.println("User deleted: " + old.getString("username"));
                    }
                }
                out.collect(value);
            }
        });
        
        env.execute("TiDB Change Processor");
    }
}

步骤4：运行Flink作业

# 编译Flink作业
mvn clean package

# 提交Flink作业
flink-1.17.0/bin/flink run -c com.example.TiDBChangeProcessor target/tidb-change-processor-1.0-SNAPSHOT.jar

# 测试数据变更
# 在TiDB中执行以下操作
INSERT INTO fgedu_users (id, username, email, age, created_at) VALUES (12, 'user12', 'user12@example.com', 33, NOW());
UPDATE fgedu_users SET age = 34 WHERE id = 12;
DELETE FROM fgedu_users WHERE id = 12;

New user added: user12
User updated: user12
User deleted: user12

# 编辑spark-defaults.conf
vi $SPARK_HOME/conf/spark-defaults.conf

# 添加以下配置
spark.sql.extensions org.apache.spark.sql.TiExtensions
spark.tispark.pd.addresses 192.168.1.10:2379,192.168.1.11:2379,192.168.1.12:2379

# 编写Spark作业
import org.apache.spark.sql.SparkSession;

public class TiDBBatchProcessor {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
            .appName("TiDB Batch Processor")
            .master("local[*]")
            .getOrCreate();
        
        // 读取TiDB数据
        spark.sql("USE fgedudb");
        spark.sql("CREATE TEMPORARY VIEW users AS SELECT * FROM fgedu_users");
        
        // 执行分析
        spark.sql("""
            SELECT 
                age, 
                COUNT(*) as user_count, 
                AVG(age) as avg_age 
            FROM users 
            GROUP BY age 
            ORDER BY age
        """).show();
        
        // 写入结果到HDFS
        spark.sql("""
            SELECT 
                age, 
                COUNT(*) as user_count, 
                AVG(age) as avg_age 
            FROM users 
            GROUP BY age 
            ORDER BY age
        """).write()
            .format("parquet")
            .save("hdfs://192.168.1.15:9000/tidb/analysis/age_stats");
        
        spark.stop();
    }
}

# 编译Spark作业
mvn clean package

# 运行Spark作业
$SPARK_HOME/bin/spark-submit --class com.example.TiDBBatchProcessor target/tidb-batch-processor-1.0-SNAPSHOT.jar

# 查看HDFS结果
bin/hdfs dfs -ls /tidb/analysis/age_stats
bin/hdfs dfs -cat /tidb/analysis/age_stats/part-00000.parquet

+---+----------+-------+
|age|user_count|avg_age|
+---+----------+-------+
| 25|         2|   25.0|
| 26|         1|   26.0|
| 34|         1|   34.0|
+---+----------+-------+