我正在尝试使用Python API for streaming在Flink上实现Kmeans Cluctering算法。我正在根据第0个索引执行key_by，然后尝试对每个组执行reduce()，以获得某种计数聚合。(CentroidAccumulator()) (2, <tuple>, count)
(3
浏览 30提问于2019-05-13得票数 1
回答已采纳
我为Flink流找到的大多数示例都会将结果发送给std-out。实际上我想把数据保存到Hadoop。我读过，理论上，你可以和Flink一起流到任何你喜欢的地方。;import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.
浏览 8提问于2015-12-28得票数 4
我希望像这个/api/resource/hSkR3V9aS一样在URL中看到散列id，而不仅仅是简单的自动增量id /api/resource/34。我的第一个想法是使用，只需从自动增量id生成hash_id，并将两个值存储在数据库中，但问题是我必须首先使用INSERT新行数据，即GET id，然后再UPDATE UPDATE hash_id字段。第二个尝试是生成hash_id (例如，sha1)，不是
浏览 2提问于2017-10-02得票数 0
回答已采纳
1-我尝试了动物园管理员和Kubernetes HA设置，结果是一样的(下面的日志来自动物园管理员HA )。以下是不成功的例子：2021-12-11 21:55:14,001 DEBUG org.apache.flink.streaming.api.graph.StreamGraphHasherV2my-ingress-1-in)-5' {id: 5, parallelism: 1, user function: }
2021-12-11 21:55:14,001 DEBUG org.apache.<e
浏览 54提问于2021-12-11得票数 0
回答已采纳
期望的最终结果是数据以10分钟的批次打包，在10分钟内对齐(即00:00 <t <= 00:10，00:10 <t <= 00:20，...，23: 50 <t <= 00:00)。第二个任务侦听这个“边界主题”，对于每个接收到的10分钟边界，从Cassandra中提取数据，执行min、max、mean、stddev等计算，并将数据和这些结果打包到定义的输出目录中。
浏览 1提问于2018-07-16得票数 0
我正在python 3中创建一个生成器，它可能产生一个或更多的值。我的问题与产生单个值的生成器有关。:只打印0，<
浏览 9提问于2022-06-20得票数 0
回答已采纳
我们有一个SQL作业(表API)，它将从Kafka主题(8个分区)中提供的内容作为源读取，并在与其他数据源进行一些聚合之后，将其汇回另一个Kafka主题，以计算最便宜的数据源，并在该结果之上聚合额外的数据，但经过一段时间后，每当数据源中发生某些事件时，它们都会意外地导致墓碑(但有时是正确设置的)结果，在检查它们之后，我们注意到不应该这样做，主要是因为对该项有实际最便宜的报价，相关的联接行确实存在。数据槽
这是同一项的data，我们可以看到，由于数据源的更新，最近的更新是同时生成的
浏览 12提问于2022-09-22得票数 1
回答已采纳
1回答
我正在运行卡夫卡和Flink作为码头集装箱在我的mac。package com.p81.datapipeline.swg;
import org.apache.flink.streaming<
浏览 2提问于2021-11-24得票数 0
回答已采纳
支持多个运行程序后端，包括Apache和Flink。我很熟悉Spark/Flink，我正试图看到Beam批量处理的利弊。从的角度来看，它觉得它非常类似于本机Spark/Flink的等价物，可能有一个稍微冗长一些的语法。
是否有更好的例子突出梁模型的其他利弊？是否有任何关于失控如何
浏览 4提问于2017-04-24得票数 108
回答已采纳
现在，这些文档是按时间间隔生成的，大约每小时生成一次。]; }}问题是它太慢了，需要我在CouchDB配置中使用reduce_limit=false。
感觉必须有一种更有效的方式来做这件事。更新相同的文档不是一个选项--历史记录很重
浏览 0提问于2012-05-01得票数 2
回答已采纳
通常我们必须(1)声明一个列表(2)，用sum()计算这个列表的和[1,5,9,13,17,21,25,29,33,37,…](2)如何快速得到该列表中第101元素到第200元素的和？
浏览 3提问于2017-02-06得票数 4
回答已采纳
，而l存储原始序列的长度。api的)等价物digitize，名为bucketize，正如Tensorflow中的digitize的this search中提到的那样。但是似乎引用的hidden_ops.txt已经从Tensorflow中删除了，我不清楚python api中是否仍然支持(也将支持) tensorflow::ops::Bucketize函数。但是这个尝试失败了，因为 seq_ids = tf.train.piecewise_constant(tf.range(tf.math.reduce_s
浏览 36提问于2019-02-20得票数 2
1回答
数组中元素和索引的和
、、、
如果我有像1，2，3，4，5这样的号码.我想用第一个元素和第一个索引来计算一个方程，然后添加到下一个带索引的元素，等等。我怎么能这么做？B = [0.183324722, 0.209108771, 0.142648498, 0.012475611, -0.075291522, 0.076201285,
浏览 3提问于2016-04-02得票数 1
回答已采纳
您能帮助我理解为什么在Python中使用filter()而不是def-function吗？只是它的可读性更好。
谢谢。
浏览 1提问于2015-12-29得票数 1
9回答
据我所知，reduce函数接受一个列表l和一个函数f。然后，它对列表的前两个元素调用函数f，然后使用下一个列表元素和前一个结果重复调用函数f。下面的函数计算阶乘。def fact(n): return 1
return fact(x) * fact(y)
print reduce(re
浏览 2提问于2012-02-02得票数 42
回答已采纳
下面是用Python3.5.1和Spark1.5.2编写的代码： """ """
A copy of the rough code from Python 2's reduceWhy did Python</e
浏览 0提问于2015-12-26得票数 15
回答已采纳
我想使用Python作为映射器和还原器，因为我最熟悉这种语言，而且它是我的同行最熟悉的语言。我觉得最简单的方法是通过Google DataProc实例来设置它，所以我也在运行它。Mapper和reducer代码来自，修改后可以使用Python3。#!版本：Python 3.7.3 (default, Mar 27 2019, 22:11:17)
[GCC 7.3.0] :: Anacon
浏览 4提问于2019-11-12得票数 2
回答已采纳
1回答
我有一个包含所有要转换的值的csv……我想读取csv文件并生成一个值文件，但当我加载并尝试调用.pot _( .pot )时，没有生成任何值。不确定为什么不能从csv生成包含动态内容的.pot文件。下面是我的代码片段：import gettext
浏览 2提问于2013-01-08得票数 0
回答已采纳
主要来自于Java的编码，并想知道Python是否可以使用条件和在它的for循环(如Java和C)中进行不同类型的增量。抱歉，如果这看起来是个简单的问题。
浏览 14提问于2020-07-01得票数 15
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
云直播
腾讯会议活动推荐
腾讯云开发者
