我需要有关AWS Kinesis Analytics功能的一些帮助。
我有一个包含这些数据的流：
hubId (Integer)
datetime (timestamp)
fid (varchar)
path (varchar)
我想将这些数据聚合到另一个流中，以hubID分组来计算每小时的行数（浏览量）和每小时的不同fid数（访客）。
目标流：
profilesite_id(Integer) = hubId from source stream
datetime (timestamp)
visitors (Integer)
pageviews (Integer)
所以在MySQL中，我的功能是这样的：
SELECT hubId, CONCAT_WS(':', SUBSTR(datetime, 1, 13), '00:00') datetime, COUNT(*) pageviews, COUNT(DISTINCT(fid)) visitors
FROM tableStream
WHERE timestamp >= CURDATE()
GROUP BY hubId, CONCAT_WS(':', SUBSTR(datetime, 1, 13), '00:00');
我试图将此请求转换为Kinesis Analytics，但是这非常困难（对我来说第一次...对不起：））。
因此，我目前拥有此Kinesis Analytics功能：
CREATE OR REPLACE STREAM "bore_agg" (profilsite_id SMALLINT, datetime TIMESTAMP, visitors INT, pageviews INT);
-- Create pump to insert into output
CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "bore_agg"
-- Select all columns from source stream
SELECT
SOURCE_SQL_STREAM_001."hubId" profilsite_id,
CHAR_TO_TIMESTAMP('yyyy-MM-DD hh:mm:ss', TIMESTAMP_TO_CHAR('YYYY-MM-DD HH:00:00', SOURCE_SQL_STREAM_001."datetime")) datetime,
COUNT(DISTINCT(SOURCE_SQL_STREAM_001."fid")) visitors,
COUNT(*) pageviews
FROM SOURCE_SQL_STREAM_001
WHERE SOURCE_SQL_STREAM_001."datetime" >= CHAR_TO_TIMESTAMP('yyyy-MM-DD hh:mm:ss', TIME_TO_CHAR('YYYY-MM-DD HH:00:00', CURRENT_TIME))
GROUP BY
SOURCE_SQL_STREAM_001."hubId",
CHAR_TO_TIMESTAMP('yyyy-MM-DD hh:mm:ss', TIMESTAMP_TO_CHAR('YYYY-MM-DD HH:00:00', SOURCE_SQL_STREAM_001."datetime"));
但是我有这个错误，我不太明白该怎么做：
您的SQL代码有错误，更新您的应用。错误信息：失败的SQL命令：创建或替换泵“ STREAM_PUMP”作为插入“ bore_agg”的选择SOURCE_SQL_STREAM_001。“ hubId” profilsite_id，CHAR_TO_TIMESTAMP（'yyyy-MM-DD hh：mm：ss'，TIMESTAMP_TO_CHAR（'YYYY-MM-DDHH：00：00'，SOURCE_SQL_STREAM_001。“ datetime”））datetime，COUNT（DISTINCT（SOURCE_SQL_STREAM_001。“ fid”））位访客，COUNT（*）位访客来自SOURCE_SQL_STREAM_001的综合浏览量SOURCE_SQL_STREAM_001。“ datetime”> = CHAR_TO_TIMESTAMP（'yyyy-MM-DDhh：mm：ss'，TIME_TO_CHAR（'YYYY-MM-DD HH：00：00'，CURRENT_TIME））GROUP BYSOURCE_SQL_STREAM_001。“ hubId”，CHAR_TO_TIMESTAMP（'yyyy-MM-DDhh：mm：ss'，TIMESTAMP_TO_CHAR（'YYYY-MM-DD HH：00：00'，SOURCE_SQL_STREAM_001。“ datetime”））。 SQL错误消息：从第9行开始第1列到第11行，第120列：无法聚合无限流：未指定GROUP BY子句或不包含任何单调性表达式。
有人能把我带到正确的方向吗？
提前感谢：）
托马斯
我知道已经有一段时间了，因为亚当在这里（唯一的）回应。因此，以防万一它对某人有所帮助，正如Adam指出的那样，如果您考虑一下，Data Analytics流可以是“无限”的输入。因此，您需要告诉停在哪里；即“汇总流中最后一分钟或一小时的数据”。因此，在此示例中（下面的代码），它将聚合流的传入数据，直到指定的分钟或小时为止。
注意：请记住，您首先需要创建一个结构相同的STREAM（返回的列数和数据类型），然后通过创建一个PUMP将“ INSERT-SELECT”运行到新流中，扫描输入数据并返回结果的过程（将在初始步骤中插入到STREAM中）。
示例：
-- ** Aggregate (COUNT, AVG, etc.) + Tumbling Time Window **
-- Performs function on the aggregate rows over a 10 second tumbling window for a specified column.
-- .----------. .----------. .----------.
-- | SOURCE | | INSERT | | DESTIN. |
-- Source-->| STREAM |-->| & SELECT |-->| STREAM |-->Destination
-- | | | (PUMP) | | |
-- '----------' '----------' '----------'
-- STREAM (in-application): a continuously updated entity that you can SELECT from and INSERT into like a TABLE
-- PUMP: an entity used to continuously 'SELECT ... FROM' a source STREAM, and INSERT SQL results into an output STREAM
-- Create output stream, which can be used to send to a destination
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM (ingest_time TIMESTAMP, vendorid int, count_vs_time int);
CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
-- Query 1):
-- Group by VendorID over the last 60 seconds of the stream.
SELECT STREAM STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND) AS ingest_time, "vendorid", COUNT(*)
FROM "SOURCE_SQL_STREAM_001"
GROUP BY STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND), "vendorid";
--Query 2)
-- Group by VendorID and count, over the last hour of the stream.
CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM (hour_range TIMESTAMP, vendorid int, count_last_hr int);
CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
SELECT STREAM FLOOR("SOURCE_SQL_STREAM_001".ROWTIME TO HOUR) AS hour_range, "vendorid", COUNT(*) as count_last_hr
FROM "SOURCE_SQL_STREAM_001"
GROUP BY FLOOR("SOURCE_SQL_STREAM_001".ROWTIME TO HOUR), "vendorid";
HTH。
卡洛斯。