如何按操作员从 Hive 组中获取元素的数组/包?
我想按给定的字段分组并获得分组字段的输出.下面是我试图实现的一个例子:- 想象一个名为“sample_table"的表，其中包含如下两列:- F1 F2001 111001 222001 123002 222002 333003 555 我想编写将提供以下输出的 Hive 查询:- 001 [111, 222, 123]002 [222, 333]003【555】 在 Pig 中，这
..
bigdata相关内容
在 spark join 中，表顺序是否像猪一样重要?
关于Spark - 加入 2 个 PairRDD 元素 在 pig 中进行常规 join 时，join 中的最后一个表不会被带入内存，而是通过流式传输，因此如果 A 每个键的基数较小而 B 的基数较大，则执行 join 明显更好A, B比join A by B，从性能角度(避免溢出和OOM) spark 中有类似的概念吗?我没有看到任何这样的建议，想知道这怎么可能?在我看来，实现与 p
..
如何从 NiFi 中的 GetFilesProcessor 读取文件
以下是我的流程: GetFile >ExecuteSparkInteractive >放置文件 我想从 ExecuteSparkInteractive 处理器中的 GetFile 处理器读取文件，应用一些转换并将其放在某个位置.下面是我的流程 我在spark处理器的code部分下写了spark scala code: val sc1=sc.textFile("local_path")s
..
如何从 NiFi 中的 GetFilesProcessor 读取文件
以下是我的流程: GetFile >ExecuteSparkInteractive >放置文件 我想从 ExecuteSparkInteractive 处理器中的 GetFile 处理器读取文件，应用一些转换并将其放在某个位置.下面是我的流程 我在spark处理器的code部分下写了spark scala code: val sc1=sc.textFile("local_path")s
..
Kafka 流 - 第一个示例 WordCount 未正确计算第一圈
我正在研究 Kafka Streams，但我对 Java 8 中 WordCount 的第一个示例有疑问，该示例取自文档. 使用最新版本的 kafka 流、Kafka Connect 和 WordCount lambda 表达式示例. 我遵循以下步骤:我在 Kafka 中创建了一个输入主题和一个输出主题.启动应用程序流，然后通过从 .txt 文件中插入一些单词来上传输入主题 在第
..
在 Kafka 流作业中进行同步数据库查询或宁静调用是一个好习惯吗?
我使用Kafka流处理实时数据，在Kafka流任务中，我需要访问MySQL查询数据，需要调用另一个restful服务. 所有操作都是同步的. 恐怕同步调用会降低流任务的处理能力. 这是一个好习惯吗?或者这样做有什么好主意吗? 解决方案 更好的方法是将 MySQL 表流式传输到 Kafka，并访问那里的数据.这具有将流应用程序与 MySQL 数据库分离的优势.如果您将来不再
..
Kafka 流 - 第一个示例 WordCount 未正确计算第一圈
我正在研究 Kafka Streams，但我对 Java 8 中 WordCount 的第一个示例有疑问，该示例取自文档. 使用最新版本的 kafka 流、Kafka Connect 和 WordCount lambda 表达式示例. 我遵循以下步骤:我在 Kafka 中创建了一个输入主题和一个输出主题.启动应用程序流，然后通过从 .txt 文件中插入一些单词来上传输入主题 在第
..
Kafka 流 - 第一个示例 WordCount 未正确计算第一圈
我正在研究 Kafka Streams，但我对 Java 8 中 WordCount 的第一个示例有疑问，该示例取自文档. 使用最新版本的 kafka 流、Kafka Connect 和 WordCount lambda 表达式示例. 我遵循以下步骤:我在 Kafka 中创建了一个输入主题和一个输出主题.启动应用程序流，然后通过从 .txt 文件中插入一些单词来上传输入主题 在第
..
KafkaSpout 元组重放抛出空指针异常
我正在使用storm 1.0.1和Kafka 0.10.0.0以及storm-kafka-client 1.0.3. 请在下面找到我的代码配置. kafkaConsumerProps.put(KafkaSpoutConfig.Consumer.KEY_DESERIALIZER, "org.apache.kafka.common.serialization.ByteArrayDeserial
..
如何清理 Kafka 0.8.2 中压缩日志中的旧段
我知道在新的 Kafka 版本中，我们有新的保留策略选项 - 日志压缩，删除具有相同密钥的旧版本消息.但是很长一段时间后，我们会得到太多带有旧消息的压缩日志段.我们如何自动清理这个压缩日志? UDPATE: 我应该澄清一下，当时我们需要紧凑的日志和清理旧消息的方法.我在这里找到了关于相同问题的讨论 http://grokbase.com/t/kafka/users/14bv6gaz0t
..
在kafka-storm中无法将偏移数据写入zookeeper
我正在设置一个风暴集群来计算实时趋势和其他统计数据，但是我在将“恢复"功能引入这个项目时遇到了一些问题，因为它允许 kafka-spout 最后读取的偏移量(kafka-spout 的源代码来自https://github.com/apache/incubator-storm/tree/master/external/storm-kafka)被记住.我以这种方式启动我的 kafka-spout:
..
为什么Kafka消费者性能很慢?
我有一个简单的主题，一个简单的 Kafka 消费者和生产者，使用默认配置. 程序很简单，我有两个线程. 在生产者中，它不断发送 16 个字节的数据. 在消费者端，它不断接收. 我发现生产者的吞吐量大约为 10MB/s，这很好. 但消费者的吞吐量仅为 0.2MB/s.我已经禁用了所有调试日志，但这并没有让它变得更好.测试在本地机器上运行.任何机构都知道出了什么问题?谢谢
..
为什么卡夫卡这么快
如果我有相同的硬件，使用 Kafka 或我们当前的解决方案(ServiceMix/Camel).有什么区别吗?Kafka 可以处理比它“更大"的数据吗?为什么? 有一篇文章说它可以有多快?但是我仍然不清楚为什么 Kafka 与其他解决方案相比如此之快?对 Apache Kafka 进行基准测试:200 万次写入每秒(在三台便宜的机器上) 解决方案 Kafka 运行速度快的原因有很多.
..
在 Kafka 流作业中进行同步数据库查询或宁静调用是一个好习惯吗?
我使用Kafka流处理实时数据，在Kafka流任务中，我需要访问MySQL查询数据，需要调用另一个restful服务. 所有操作都是同步的. 恐怕同步调用会降低流任务的处理能力. 这是一个好习惯吗?或者这样做有什么好主意吗? 解决方案 更好的方法是将 MySQL 表流式传输到 Kafka，并访问那里的数据.这具有将流应用程序与 MySQL 数据库分离的优势.如果您将来不再
..
kappa-architecture 和 lambda-architecture 有什么区别
如果 Kappa-Architecture 直接对流进行分析而不是将数据拆分为两个流，那么在像 Kafka 这样的消息系统中，数据存储在哪里?还是可以在数据库中重新计算? 单独的批处理层是否比使用流处理引擎重新计算进行批处理更快? 解决方案 “要考虑的一个非常简单的情况是算法应用于实时数据和历史数据是一致的.那么就是使用相同的代码库来处理显然非常有益历史和实时数据，从而实现用例使
..
每个生产者的 Kafka 主题
假设我有多个设备.每个设备都有不同类型的传感器.现在我想将每个传感器的每个设备的数据发送到 kafka.但我对 kafka 主题感到困惑.用于处理这些实时数据 每个设备都有一个 kafka 主题，并且该设备的所有传感器都将数据发送到特定的 kafka 主题，或者我应该创建一个主题并让所有设备将数据发送到该主题. 如果我采用第一种情况，我们将为每个设备创建主题，然后， 设备 1(传
..
Flink中的水印和触发器有什么区别?
我读到，“..排序运算符必须缓冲它收到的所有元素.然后，当它收到水印时，它可以对时间戳低于水印的所有元素进行排序，并按排序顺序发出它们.这是正确的，因为水印表明不会有更多元素可以到达，这些元素将与已排序的元素混合......" - https://cwiki.apache.org/confluence/display/FLINK/Time+and+Order+in+Streams 因此，水印
..
在 Flink 流中使用静态 DataSet 丰富 DataStream
我正在编写一个 Flink 流程序，其中我需要使用一些静态数据集(信息库，IB)来丰富用户事件的 DataStream. 例如假设我们有一个买家的静态数据集，我们有一个传入的事件点击流，对于每个事件，我们想要添加一个布尔标志，指示事件的执行者是否是买家. 实现此目的的理想方法是按用户 ID 对传入流进行分区，让数据集中的买方设置再次按用户 ID 进行分区，然后在此数据集中查找流中的每个
..
Python + Beam + Flink
我一直在尝试让 Apache Beam 可移植性框架与 Python 和 Apache Flink 一起工作，但我似乎找不到一套完整的说明来使环境正常工作.是否有任何参考资料包含使简单的 Python 管道正常工作的先决条件和步骤的完整列表? 解决方案 总体而言，对于本地便携式运行器 (ULR)，参见维基，引用自那里: 运行 Python-SDK 管道: 将容器编译为本地构建:
..
Flink:Jobmanager UI 中设置的并行度与任务槽有什么关系?
假设我有 8 个任务管理器和 16 个任务槽.如果我使用 Jobmanager UI 提交作业并将并行度设置为 8，我是否只使用 8 个任务槽? 如果我有 8 个有 8 个插槽的任务管理器，并以 8 的并行度提交同一个作业怎么办?它是完全一样的吗?还是数据处理方式有差异? 谢谢. 解决方案 Flink 集群中的任务槽总数定义了最大并行度，但使用的槽数可能会超过实际并行度.例如，
..