在 Kinesis Data Analytics for Java Applications 中使用操作符转换数据 - Amazon Kinesis Data Analytics
AWS 文档中描述的 AWS 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 AWS 服务入门

在 Kinesis Data Analytics for Java Applications 中使用操作符转换数据

要在Kinesis Data Analytics for Java application中转换传入数据,您可以使用 Apache Flink 操作符。Apache Flink 操作符将一个或多个数据流转换为新的数据流。新数据流包含来自原始数据流的修改的数据。Apache Flink 提供超过 25 个预构建的流处理操作符。有关更多信息,请参阅 Apache Flink 文档中的操作符

本主题包含下列部分:

转换操作符

以下是对 JSON 数据流的某个字段进行简单文本转换的示例。

该代码创建转换的数据流。新数据流具有与原始流相同的数据,并在 TICKER 字段内容后面附加“ Company”字符串。

DataStream<ObjectNode> output = input.map( new MapFunction<ObjectNode, ObjectNode>() { @Override public ObjectNode map(ObjectNode value) throws Exception { return value.put("TICKER", value.get("TICKER").asText() + " Company"); } } );

聚合操作符

以下是一个聚合操作符示例。该代码创建聚合的数据流。该操作符创建一个 5 秒的滚动窗口,并返回窗口中具有相同 TICKER 值的记录的 PRICE 值之和。

DataStream<ObjectNode> output = input.keyBy(node -> node.get("TICKER").asText()) .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) .reduce((node1, node2) -> { double priceTotal = node1.get("PRICE").asDouble() + node2.get("PRICE").asDouble(); node1.replace("PRICE", JsonNodeFactory.instance.numberNode(priceTotal)); return node1; });

有关使用操作符的完整代码示例,请参阅入门。在 Kinesis Data Analytics Java 示例 GitHub 存储库的入门中提供了入门应用程序的源代码。