darenwkt commented on PR #211:
URL:
https://github.com/apache/flink-connector-aws/pull/211#issuecomment-3005842462
Thanks for the PR, looks good to me and have approved.
Ran some quick test to verify it works with Flink 2.0
### Kinesis Source/Sink
- Set up
```
% ./flink-2.0.0/bin/start-cluster.sh
% ./flink-2.0.0/bin/sql-client.sh --jar
/Users/darenwkt/Documents/github/flink-connector-dynamodb/flink-connector-aws/flink-sql-connector-aws-kinesis-streams/target/flink-sql-connector-aws-kinesis-streams-5.1-SNAPSHOT.jar
```
- On SQL client:
- Create Tables
```
CREATE TABLE random_data (
`user_id` BIGINT,
`item_id` BIGINT,
`category_id` BIGINT,
`behavior` STRING,
`ts` TIMESTAMP(3)
) WITH (
'connector' = 'datagen',
'rows-per-second' = '10'
);
CREATE TABLE KinesisTable (
`user_id` BIGINT,
`item_id` BIGINT,
`category_id` BIGINT,
`behavior` STRING,
`ts` TIMESTAMP(3)
)
PARTITIONED BY (user_id, item_id)
WITH (
'connector' = 'kinesis',
'stream.arn' =
'arn:aws:kinesis:us-east-1:redacted:stream/ExampleInputStream',
'aws.region' = 'us-east-1',
'source.init.position' = 'LATEST',
'format' = 'csv'
);
```
- Insert data into SInk
```
INSERT INTO KinesisTable SELECT * FROM random_data;
```
- Read data into Source
```
SELECT * FROM KinesisTable;
```
- Verified data is submitted to sink kinesis stream, and read the stream
source to verify data is sent
### Firehose Sink
- Set up
```
% ./flink-2.0.0/bin/start-cluster.sh
% ./flink-2.0.0/bin/sql-client.sh --jar
/Users/darenwkt/Documents/github/flink-connector-dynamodb/flink-connector-aws/flink-sql-connector-aws-kinesis-firehose/target/flink-sql-connector-aws-kinesis-firehose-5.1-SNAPSHOT.jar
```
- On SQL client:
- Create Tables
```
CREATE TABLE random_data (
`user_id` BIGINT,
`item_id` BIGINT,
`category_id` BIGINT,
`behavior` STRING,
`ts` TIMESTAMP(3)
) WITH (
'connector' = 'datagen',
'rows-per-second' = '10'
);
CREATE TABLE FirehoseTable (
`user_id` BIGINT,
`item_id` BIGINT,
`category_id` BIGINT,
`behavior` STRING
)
WITH (
'connector' = 'firehose',
'delivery-stream' = 'user_behavior',
'aws.region' = 'us-east-2',
'format' = 'csv'
);
```
- Insert data into SInk
```
INSERT INTO FirehoseTable SELECT * FROM random_data;
```
- Verify data appears in Firehose delivery and is found in destination S3:
https://analytics-dev-test-redacted.s3.us-east-1.amazonaws.com/2025/06/25/18/analyticsTestOutput-3-2025-06-25-18-33-58-146eb048-00d8-4977-9b7b-e23a17c33963
### DDB Sink
- Set up
```
% ./flink-2.0.0/bin/start-cluster.sh
% ./flink-2.0.0/bin/sql-client.sh --jar
flink-sql-connector-dynamodb-5.1-SNAPSHOT.jar
```
- On SQL client:
- Create Tables
```
CREATE TABLE random_data (
`user_id` STRING,
`item_id` BIGINT,
`category_id` BIGINT,
`behavior` STRING,
`ts` TIMESTAMP(3)
) WITH (
'connector' = 'datagen',
'rows-per-second' = '10'
);
CREATE TABLE DynamoDbTable (
`user_id` STRING,
`item_id` BIGINT,
`category_id` BIGINT,
`behavior` STRING,
`ts` TIMESTAMP(3)
) PARTITIONED BY ( user_id )
WITH (
'connector' = 'dynamodb',
'table-name' = 'ddbConnectorTest',
'aws.region' = 'us-east-1'
);
```
- Insert data into SInk
```
INSERT INTO DynamoDbTable SELECT * FROM random_data;
```
- Verify data appears in DDB
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
