使用 eventbridge 管道将动态数据插入到 redshift

Question

我一直在使用 aws eventbridge 创建一个具有 kinesis 源的管道，源、过滤和丰富方面一切都很好。但是当我尝试将信息发送到红移时，它就不起作用。日志显示一切正常，但新数据未在 redshift 中显示。

我使用的查询是

INSERT INTO public.event_bridge_test (id, status, test, test1) VALUES ('11234124321', '13123123' ,'test', $.data);

数据只有一个“hello world！”要测试的字符串。此插入不起作用，但如果我使用类似

INSERT INTO public.event_bridge_test (id, status, test, test1) VALUES ('11234124321', '13123123' ,'test', 'test1');

的内容，则插入新行时不会出现红移问题。 我已经阅读了几个小时的文档，甚至尝试过 aws ia，但没有成功。还尝试使用目标输入变压器获得相同的结果。

我尝试过的所有内容的列表：

INSERT INTO public.event_bridge_test (id, status, test, test1) VALUES ('5', json_extract_path(to_json('$context.event.detail'), '$.data.status'), 'test', 'test1')
INSERT INTO public.event_bridge_test (id, status, test, test1) VALUES ('6', $.data.status, 'test', 'test1')
INSERT INTO public.event_bridge_test (id, status, test, test1) VALUES ('7', $.status, 'test', 'test1')
INSERT INTO public.event_bridge_test (id, status, test, test1) VALUES ('8', json_extract_path(to_json('$context.event.detail'), '$.status'), 'test', 'test1')
INSERT INTO public.event_bridge_test (id, status, test, test1) VALUES ('9', json_extract_path(to_json('event.detail'), '$.status'),to_json('event.detail'), 'test', 'test1')
INSERT INTO public.event_bridge_test (id, status, test, test1) VALUES ('10', JSON_EXTRACT_PATH_TEXT('$context.event.detail','data', 'status'), 'test', 'test1')
INSERT INTO public.event_bridge_test (id, status, test, test1) VALUES ('11', json_extract_path(input, 'status') ,'test', 'test1');
INSERT INTO public.event_bridge_test (id, status, test, test1) VALUES ('11234124321', '13123123' ,'test', TO_CHAR($));

Answer 1

这听起来可能是一个 COMMIT 问题（一切正常，但没有数据显示）。数据库不会永久存储没有 COMMIT 的更改，并且不允许另一个会话查看已提交的更改，直到该会话通过 COMMIT 或 ROLLBACK 更新为止。

所以这引出了几个问题。

• 您正在查看哪些日志？
• 如何检查数据更改是否不在数据库中？
• 您能在 redshift 目录表（如 stl_querytext）中看到您发出的 SQL 吗？
• 您如何插入有效的数据集？使用工作台会话？
• 您要发送哪些 COMMIT，以及与数据库的连接有哪些 COMMIT 设置？

顺便说一句，由于您使用的是 Kinesis，因此您很有可能希望该解决方案能够扩展到高容量。您将遇到的问题是，以这种方式将数据插入 Redshift 的性能非常低下，并且实际上会给集群带来过大的压力，从而导致其他工作减慢。您应该考虑使用像 firehose 一样的 COPY 对数据进行批处理并通过 S3 进行加载。