使用AWS Redshift从AWS MSK中读取数据

Amazon Redshift 流式摄取的目的是简化将流式数据直接从流式服务摄取到 Amazon Redshift 或 Amazon Redshift Serverless 的过程。

官方文档[1]中有详细步骤。用unauthenticated, IAM 的方式均可以进行连接，只不过使用的是不同端口：9092或者9098

[1] https://docs.amazonaws.cn/redshift/latest/dg/materialized-view-streaming-ingestion-getting-started-MSK.html

使用IAM方式：

dev=# CREATE EXTERNAL SCHEMA external_schema_name_msk
FROM MSK
IAM_ROLE default
AUTHENTICATION iam
URI 'b-3.testtest.6ruea7.c2.kafka.cn-northwest-1.amazonaws.com.cn:9098';
CREATE SCHEMA
Time: 309.537 msdev=# SELECT schemaname, tablename
FROM SVV_EXTERNAL_TABLES
WHERE schemaname = 'external_schema_name_msk';schemaname        |      tablename
--------------------------+---------------------external_schema_name_msk | __amazon_msk_canaryexternal_schema_name_msk | __consumer_offsetsexternal_schema_name_msk | kafka-default-topic
(3 rows)dev=# CREATE MATERIALIZED VIEW common_msk_cdp_wecom_tag AUTO REFRESH YES AS
SELECT * FROM external_schema_name_msk."kafka-default-topic";
CREATE MATERIALIZED VIEW
Time: 667.501 msdev=# select kafka_timestamp from common_msk_cdp_wecom_tag;kafka_timestamp 
-----------------
(0 rows)Time: 18.564 msdev=# refresh MATERIALIZED VIEW common_msk_cdp_wecom_tag;
INFO:  Materialized view common_msk_cdp_wecom_tag was incrementally updated successfully. Stream returned no new data.
REFRESH
Time: 1493.006 msdev=# select kafka_timestamp from common_msk_cdp_wecom_tag;kafka_timestamp
-------------------------2024-10-12 04:49:19.6932024-10-12 04:49:20.1362024-10-12 04:49:20.1932024-10-12 04:49:20.2062024-10-12 04:49:20.2142024-10-12 04:49:20.2412024-10-12 04:49:20.2482024-10-12 04:49:20.256
(8 rows)Time: 6141.639 ms

使用Unauthenticated 方式：

dev=# CREATE EXTERNAL SCHEMA msktest
dev-# FROM MSK
dev-# IAM_ROLE 'arn:aws-cn:iam::123456789:role/shiranredshift'
dev-# AUTHENTICATION none
dev-# CLUSTER_ARN 'arn:aws-cn:kafka:cn-north-1:123456789:cluster/shiranmsk/03260cbd-b23c-442f-a858-8af96e6b90de-2';
CREATE SCHEMA
dev=#
dev=#
dev=# CREATE MATERIALIZED VIEW mskview AUTO REFRESH YES AS
dev-# SELECT *
dev-# FROM msktest."mirror1";
CREATE MATERIALIZED VIEW
dev=#
dev=#
dev=# select * from mskview;kafka_partition | kafka_offset | kafka_timestamp_type | kafka_timestamp | kafka_key | kafka_value | kafka_head
ers | refresh_time
-----------------+--------------+----------------------+-----------------+-----------+-------------+-----------
----+--------------
(0 rows)dev=#