使用 Amazon Q Developer CLI 快速搭建各种场景的 Flink 数据同步管道

sudo cp /home/hadoop/flink-sql-connector-mysql-cdc-2.3.0.jar /usr/lib/flink/lib/
sudo chown flink:flink flink-sql-connector-mysql-cdc-2.3.0.jar
sudo chmod 755 flink-sql-connector-mysql-cdc-2.3.0.jar

cd /usr/lib/flink/bin
./stop-cluster.sh
./start-cluster.sh

kafka-topics.sh --create --bootstrap-server b-3.pingaws.jsi6j6.c6.kafka.us-east-1.amazonaws.com:9092,b-1.pingaws.jsi6j6.c6.kafka.us-east-1.amazonaws.com:9092,b-2.pingaws.jsi6j6.c6.kafka.us-east-1.amazonaws.com:9092 --replication-factor 3 --partitions 1 --topic flink-cdc-kafka

./yarn-session.sh -d -s 1 -jm 1024 -tm 2048 -nm flink-cdc-kafka

yarn application -list

Total number of applications (application-types: [], states: [SUBMITTED, ACCEPTED, RUNNING] and tags: []):1
                Application-Id	    Application-Name	    Application-Type	      User	     Queue	             State	       Final-State	       Progress	                       Tracking-URL
application_1687084195703_0001	     flink-cdc-kafka	        Apache Flink	    hadoop	   default	           RUNNING	         UNDEFINED	           100%	         http://172.31.88.126:46397

$ ./sql-client.sh embedded -s flink-cdc-kafka

SET execution.checkpointing.interval = 3s;

CREATE TABLE products (
    id INT,
    name STRING,
    description STRING,
    PRIMARY KEY (id) NOT ENFORCED
  ) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'Source RDS endpoint',
    'port' = '3306',
    'username' = 'user',
    'password' = 'password',
    'database-name' = 'mydb',
    'table-name' = 'products'
  );

CREATE TABLE enriched_orders (
   order_id INT,
   order_date TIMESTAMP(0),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   order_status BOOLEAN,
   product_name STRING,
   product_description STRING,
   shipment_id INT,
   origin STRING,
   destination STRING,
   is_arrived BOOLEAN,
   PRIMARY KEY (order_id) NOT ENFORCED
 ) WITH (
 'connector' = 'kafka',
 'topic' = 'flink-cdc-kafka',
 'properties.bootstrap.servers' = ‘xxxx:9092',
 'properties.group.id' = 'flink-cdc-kafka-group',
 'scan.startup.mode' = 'earliest-offset',
 'format' = 'debezium-json'
 );

INSERT INTO enriched_orders
 SELECT o.*, p.name, p.description, s.shipment_id, s.origin, s.destination, s.is_arrived
 FROM orders AS o
 LEFT JOIN products AS p ON o.product_id = p.id
 LEFT JOIN shipments AS s ON o.order_id = s.order_id;

kafka-console-consumer.sh --bootstrap-server xxt:9092 --topic multi-table-join --from-beginning
{"before":null,"after":{"order_id":10001,"order_date":"2020-07-30 10:08:22","customer_name":"Jark","price":50.5,"product_id":102,"order_status":false,"product_name":"car battery","product_description":"12V car battery","shipment_id":10001,"origin":"Beijing","destination":"Shanghai","is_arrived":false},"op":"c"}

CREATE TABLE enriched_orders_output(
   order_id INT,
   order_date TIMESTAMP(0),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   order_status BOOLEAN,
   product_name STRING,
   product_description STRING,
   shipment_id INT,
   origin STRING,
   destination STRING,
   is_arrived BOOLEAN,
   PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
 'connector' = 'jdbc',
 'url' = 'jdbc:mysql://<Target RDS endpoint>:3306/mydb',
 'table-name' = 'kafka_orders_output', 
 'username' = ‘user', 
 'password' = ‘password'
);

INSERT INTO enriched_orders_output SELECT * FROM enriched_orders;

mysql> INSERT INTO orders
    -> VALUES (default, '2020-07-30 15:22:00', 'Jark', 29.71, 104, false); 
mysql> INSERT INTO shipments
    -> VALUES (default,10004,'Shanghai','Beijing',false);
mysql> UPDATE orders SET order_status = true WHERE order_id = 10004;
mysql> UPDATE shipments SET is_arrived = true WHERE shipment_id = 1004;

SET execution.checkpointing.interval = 3s; 

[
  {
    "Classification": "yarn-site",
    "Properties": {
      "yarn.resourcemanager.am.max-attempts": "10"
    }
  },
  {
    "Classification": "flink-conf",
    "Properties": {
      "high-availability": "zookeeper",
      "high-availability.storageDir": "hdfs:///user/flink/recovery",
      "high-availability.zookeeper.path.root": "/flink",
      "high-availability.zookeeper.quorum": "%{hiera('hadoop::zk')}",
      "yarn.application-attempts": "10"
    }
  }
]

flink-connector-jdbc-1.16.2.jar
flink-sql-connector-kafka-1.16.1.jar
flink-sql-connector-mysql-cdc-2.3.0.jar
flink-sql-connector-dynamodb-4.1.0-1.16.ja
cd /usr/lib/flink/lib
sudo chown flink:flink flink-sql-connector-dynamodb-4.1.0-1.16.jar
sudo chmod 755 flink-sql-connector-dynamodb-4.1.0-1.16.jar

/usr/lib/flink/bin
sudo ./stop-cluster.sh
sudo ./start-cluster.sh

./yarn-session.sh -d -s 1 -jm 1024 -tm 2048 -nm flink-cdc-ddb

yarn application -list
                Application-Id	    Application-Name	    Application-Type	      User	     Queue	             State	       Final-State	       Progress	                       Tracking-URL
application_1739533554641_0006	       flink-cdc-ddb	        Apache Flink	    hadoop	   default	           RUNNING	         UNDEFINED	           100%	         http://172.31.86.244:45475

./sql-client.sh embedded -s flink-cdc-ddb

SET execution.checkpointing.interval = 3s;

CREATE TABLE product_view_source (
`id` int,
`user_id` int,
`product_id` int,
`server_id` int,
`duration` int,
`times` string,
`time` timestamp,
PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = ‘xxxx',
'port' = '3306',
'username' = ‘user',
'password' = ‘password',
'database-name' = 'test',
'table-name' = 'product_view'
);

CREATE TABLE product_view_kafka_sink(
`id` int,
`user_id` int,
`product_id` int,
`server_id` int,
`duration` int,
`times` string,
`time` timestamp,
PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
 'connector' = 'kafka',
 'topic' = 'flink-cdc-kafka',
 'properties.bootstrap.servers' = ‘xxxx:9092',
 'properties.group.id' = 'flink-cdc-kafka-group',
 'scan.startup.mode' = 'earliest-offset',
 'format' = 'debezium-json’
);

insert into product_view_kafka_sink select * from product_view_source;

CREATE TABLE kafka_output(
`id` int,
`user_id` int,
`product_id` int,
`server_id` int,
`duration` int,
`times` string,
`time` timestamp
) PARTITIONED BY ( id )
WITH (
  'connector' = 'dynamodb',
  'table-name' = 'kafka_ddb',
  'aws.region' = 'us-east-1’
);

INSERT INTO kafka_output SELECT * FROM product_view_kafka_sink;

{"before":null,"after":{"id":1,"user_id":1,"product_id":1,"server_id":1,"duration":120,"times":"120","time":"1587734040000"},"op":"c"}
{"before":null,"after":{"id":2,"user_id":1,"product_id":1,"server_id":1,"duration":120,"times":"120","time":"1587734040000"},"op":"c"}
…
{"before":{"id":10,"user_id":8,"product_id":1,"server_id":2,"duration":120,"times":"120","time":"1589375640000"},"after":null,"op":"d"}
{"before":null,"after":{"id":10,"user_id":8,"product_id":10,"server_id":2,"duration":120,"times":"120","time":"1589375640000"},"op":"c"}
{"before":{"id":10,"user_id":8,"product_id":10,"server_id":2,"duration":120,"times":"120","time":"1589375640000"},"after":null,"op":"d"}
{"before":null,"after":{"id":10,"user_id":8,"product_id":1,"server_id":2,"duration":120,"times":"120","time":"1589375640000"},"op":"c"}

Design Dynamodb as RDS Mysql table for chat message. Refer to SQL:
CREATE TABLE message (
  msgid STRING PRIMARY KEY,
  sessionid STRING NOT NULL,
  sender_id STRING NOT NULL,
  receiver_id STRING NOT NULL,
  message_type VARCHAR(20),
  content VARCHAR(1000),
  status VARCHAR(10),
  created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME
);
ID should be unique but incremental, considering snowflake algorithm. Each chat session(sessionid) may contain several messages with "msgid". Need query:
1. Chat sessions to get messages in time range. 
2. Recent messages for sender and receiver.
Insert 10000 rows data.
RDS: <Your RDS Mysql endpoint>, user: admin, password: xxxx

## MySQL Table Design
I've created a MySQL table with the following schema:
sql
CREATE TABLE message (
  msgid VARCHAR(255) PRIMARY KEY,
  sessionid VARCHAR(255) NOT NULL,
  sender_id VARCHAR(255) NOT NULL,
  receiver_id VARCHAR(255) NOT NULL,
  message_type VARCHAR(20),
  content VARCHAR(1000),
  status VARCHAR(10),
  created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME
);
 With indexes for common query patterns:
sql
CREATE INDEX idx_sessionid ON message(sessionid);
CREATE INDEX idx_sender_id ON message(sender_id);
CREATE INDEX idx_receiver_id ON message(receiver_id);
CREATE INDEX idx_created_at ON message(created_at);

## DynamoDB Design
 
• **Primary Key**:
  • Partition Key: PK (String) - Composite key with different prefixes
  • Sort Key: SK (String) - Composite key for sorting and querying
 
• **Global Secondary Indexes**:
  • GSI1: For session-based queries
  • GSI2: For sender-based queries
 
This design efficiently supports your required query patterns:
1. Get messages in a session within a time range
2. Get recent messages for a sender or receiver
 
## ID Generation
 
I've implemented a Snowflake-like ID generation algorithm in snowflake_id.py that creates:
• Time-sortable IDs (41 bits for timestamp)
• Unique across distributed systems (10 bits for machine ID)
• High throughput (12 bits for sequence number)

1. create_mysql_table.sql: SQL script to create the MySQL table
2. snowflake_id.py: ID generation utility
3. create_dynamodb_table.py: Script to create the DynamoDB table
4. generate_sample_data.py: Script to generate 10,000 sample messages in MySQL
5. migrate_to_dynamodb.py: Script to migrate data from MySQL to DynamoDB
6. query_examples.py: Examples of common query patterns
7. requirements.txt: Required Python packages
8. run.sh: Shell script to run the entire migration process

        SELECT * FROM message
        WHERE sessionid = %s
        AND created_at BETWEEN %s AND %s
        ORDER BY created_at ASC
        LIMIT %s

        response = self.table.query(
            IndexName='SessionTimeIndex',
            KeyConditionExpression=Key('sessionid').eq(session_id) &
                                  Key('created_at').between(start_timestamp, end_timestamp),
            Limit=limit,
            ScanIndexForward=True  # True for ascending (oldest first)
        )

Need migrate database from Couchbase to Dynamodb. Create data pipeline by Flink. I have local Couchbase 7.6.6, Flink 1.6.2. Create project which can run in intellij idea IDE

org.apache.flink.api.common.eventtime.WatermarkStrategy;

Let me fix those import errors. The issue is that some of the dependencies might not be correctly specified in the pom.xml file. Let‘s update the pom.xml to include the correct dependencies:

Exception in thread "main" java.lang.IllegalStateException: Detected an UNBOUNDED source with the 'execution.runtime-mode' set to 'BATCH'.
This error occurs because the Flink job is configured to run in BATCH mode, but your CouchbaseSource is being treated as an unbounded source (which is not allowed in BATCH mode).
Let me fix this issue by modifying the CouchbaseSource class to implement the SourceFunction interface correctly for batch processing:

Caused by: com.couchbase.client.core.error.IndexFailureException: com.couchbase.client.core.error.IndexFailureException: The server reported an issue with the underlying index 
I see the error. The issue is that the Couchbase query is trying to access a bucket named "default" but your configuration is set to use"gamesim-sample" as the bucket. Let me fix this mismatch

Use EMR flink to merge and sync Dynamodb tables.
In us-east-1 region, 3 Dynamodb tables already exist(GameMailbox1, GameMailbox2, GameMailbox3) with the same schema for gaming mail box. Sample data:
{JSON}
Field "reward" data type is map.
Each table has almost 10000 items.
Combine the 3 tables into one merged table(GameMailboxMerged) with data format like:
(JSON}
The target merged table has the same gameId. Partition key is built as "gameId#mailId". 
Data is processed via Flink includeing full load and cdc without interruption. Source table stream should be enabled. You can use existing EMR cluster "mvdemo" with flink 1.16. Make sure Flink keeps running for cdc. Use Flink datastream connector with Java for cdc. Also Java for full load.

public void invoke(GameMailboxMergedItem value, Context context) {
        try {
            // Check if this is a delete operation
            if (value.isDeleteOperation()) {
                // Delete the item from the target table
                deleteItem(value);
            } else {
                // Convert the item to a DynamoDB item and write it
                Map<String, AttributeValue> item = convertToAttributeValues(value);
                
                // Write to DynamoDB
                PutItemRequest putItemRequest = PutItemRequest.builder()
                        .tableName(tableName)
                        .item(item)
                        .build();
                
                dynamoDbClient.putItem(putItemRequest);
                LOG.info("Successfully wrote item with PK: {} to table: {}", value.getPK(), tableName);
            }

 CREATE TABLE s3_duf_3d(oneid string, duf_3d_feas_map string,
PRIMARY KEY (oneid) NOT ENFORCED
) WITH (
 'connector' = 'filesystem',
 'path' = 's3://pingawsdemo/duf_1_3d_by_region/',
 'format' = 'csv',
 'csv.field-delimiter'='\t'
);

CREATE TABLE ddb_fea (
  oneid string, type string, feas string
)
WITH (
  'connector' = 'dynamodb',
  'table-name' = 'mydemo',
  'aws.region' = 'us-east-1'
);

INSERT INTO ddb_fea
SELECT oneid, 'duf_3d' AS type, duf_3d_feas_map AS feas FROM s3_duf_3d;