1回答
可以使用Apache Beam框架输出(E1，E2，E3，E4)吗？我的目标是在给定匹配的元素对的情况下(并行)输出所有连接的组件。
浏览 7提问于2018-09-10得票数 0
背景:我正在使用TFX管道，并使用Flink作为光束的运行器(使用flink-on-K8s-operator的flink会话集群)。Flink集群有2个任务管理器，每个任务管理器有16个核心，并行度设置为32。TFX组件调用beam.io.ReadFromTFRecord来加载数据，传入一个glob文件模式。当我尝试运行该组件时，对所有160个文件的处理在Flink中的单个子任务中结束，即并行度实际上为1。如下图所示： ? ? 我尝试了所有的波
浏览 18提问于2021-09-20得票数 0
回答已采纳
我使用了一个稍作修改的wordcount示例(https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/wordcount.py723 但是，实际结果如下所示，这表明该步骤没有并行运行： PARALLEL START? 447PARALLEL STA
浏览 12提问于2020-06-25得票数 3
回答已采纳
令人困惑的是，每个关于数据流的谷歌文档都说它现在是基于Apache光束的，并将我引导到光束网站。此外，如果我查找github项目，我会发现google dataflow项目是空的，并且所有的项目都转到apache see repo。假设我现在需要创建一个管道，根据我从Apache光束中读到的，我会这样做：from apache_beam.options.pipeline_options然而，如果我使用google-cloud-dataflow，我会得到错误：no mod
浏览 1提问于2017-06-16得票数 3
是否可以读取整个文件(而不是逐行读取)？ 例如，我想读取多行JSON，我的想法是逐个文件读取，从每个文件中提取数据并从列表中创建PCollection。 这是一个好主意，还是将源JSON预处理为一个JSON文件更好，每行都是单独的JSON？ 谢谢你的预支。
浏览 17提问于2019-10-18得票数 1
回答已采纳
2回答
= [ (1, "Sangeeta,24,Kolkata"), (2, "Akshay,26,Delhi"), (3, "Sahil,26,Kolkata")]
很抱歉提出了这样一个基本的问题，但我对Apache
浏览 4提问于2022-03-01得票数 0
回答已采纳
1回答
如何在中创建任务之间的依赖关系
、、、、
我是apache的新手，并探索apache数据流的python版本。我希望按一定顺序执行数据流任务，但它以并行模式执行所有任务。如何在中创建任务依赖关系？示例代码：(在下面的代码中，sample.json文件包含5行)import logging
class Sampl
浏览 0提问于2018-03-17得票数 0
1回答
在整个管道中，没有并行性，文件中只有一行字符串。你能告诉我我做错了什么吗？我应该如何更新代码以实现上面所述的并行性？package main
"context" "time"
"github.com&
浏览 1提问于2019-03-11得票数 1
我试着在读完csv和写了几个转换之后，尝试利用写到pubsbub，但是，一旦我到了需要开始使用pubsub的地步，我就无法导入。from beam_utils.sources import CsvFileSourcefrom apache_beam.ioimport WriteToText
from apache_beam.io.gcp.pubsub import Pub
浏览 0提问于2018-08-25得票数 0
我正在使用spark runner进行此模拟： PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();
}));result.waitUntilFinish(); 我正在使用--runner=SparkRunner --sparkMaster=local[8]运行，但在重新洗牌后只使用了一个线程。
浏览 14提问于2020-08-25得票数 1
回答已采纳
我使用的是运行在Docker上的Flink 1.2.1，任务管理器作为Docker群的一部分分布在不同的am上。使用Flink Web UI上传Apache光束应用程序并尝试在作业提交点设置并行度不起作用。两者都不会使用Flink CLI提交作业。
看起来在客户机级并行度并没有被接受，它最终默认为1。当我以编程方式在Apache Beam代码中设置并行度时，它可以工作：flinkPipelineOptions.setParalleli
浏览 2提问于2017-10-23得票数 0
我是第一次使用Apache光束和Dataflow。我想使用数据集作为将使用数据流并行部署的函数的输入。这是我到目前为止所知道的：import apache_beam as beam
from apache_beam.options.pipeline_options import SetupOptionsfrom apache_beam.options.pipe
浏览 4提问于2018-11-21得票数 2
1回答
files' >> MatchFiles(input_dir + "*.csv") | 'Parse contents' >> beam.ParDo(FileToRowsFn(), additional_side_inputs)
| 'Compute average' >> beam.CombinePerKey(AverageCalculatorF
浏览 2提问于2021-08-17得票数 2
回答已采纳
1回答
我想检查一下，如果我们在apache beam管道中读取的CSV文件在执行任何转换之前满足了我期望它在Ex中的格式:字段检查、类型检查、空值检查等等。在管道外对每个文件执行这些检查将消除并行性的概念，所以我只想知道是否有可能在管道中执行它。代码可能是什么样子的示例： def process(&q
浏览 1提问于2022-04-16得票数 0
回答已采纳
当我试图在Flink运行器上运行TFX管道/Apache光束作业时，当使用1个任务管理器(在一个节点上)并行度为2(每个任务管理器2个任务槽)时，它工作得很好。但当我在多个任务管理器上以更高的并行性尝试它时，当消息在两个任务管理器上不断重复时，它会挂起： INFO org.apache.beam.runners.fnexecution.environment.ExternalEnvironmentFactory因此，我在配置时使用该地址： f"--
浏览 38提问于2021-08-10得票数 0
回答已采纳
4回答
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;谢谢。
浏览 13提问于2017-03-27得票数 1
回答已采纳
目前，我在Python中有一个apache-beam管道，在该管道中，我正在读取拼花，将其转换为dataframe来进行一些熊猫的清理，然后将其转换回我想要编写文件的parquet。看起来是这样的： | 'Read' >>beam.io.ReadFromParquetBatched(known_args.input) \
浏览 8提问于2022-10-24得票数 2
回答已采纳
2回答
我正在尝试为Apache开发一个自定义I/O连接器，它是用Python编写的。根据，在我的例子中，可拆分的DoFn (SDF)是可选择的框架。import loggingfr
浏览 28提问于2022-01-04得票数 0
对每个文件进行一系列调用以对其进行索引(例如name = blob.name，size = blob.size)
+我所能做的： return pipeline | "LOAD" >> beam</e
浏览 0提问于2018-05-22得票数 2
1回答
我创建了一个使用Apache的管道，如下所示： def process(self," >> beam.ParDo(LemmaText()) \
| 'WriteToBigQuery' >> beam.io.WriteToBigQuery('mybq', custom_gcs_temp_location=&quo
浏览 16提问于2022-01-05得票数 1
云服务器
ICP备案
对象存储
云直播
腾讯会议
腾讯云开发者
