标签: apache-beam

有没有办法检查 PCollection 是否为空？

我在 Dataflow 和 Apache Beam 的文档中没有找到任何相关内容。

我有一个简单的任务。我有一堆文件( ~100GB in total )，每一行代表一个实体。我必须将此实体发送到 JanusGraph 服务器。

2018-07-07_05_10_46-8497016571919684639 <- job id

一段时间后，我遇到 OOM，日志显示 Java 被杀死。

从数据流视图中，我可以看到以下日志：

Workflow failed. Causes: S01:TextIO.Read/Read+ParDo(Anonymous)+ParDo(JanusVertexConsumer) failed., A work item was attempted 4 times without success. Each time the worker eventually lost contact with the service. The work item was attempted on:

从堆栈驱动程序视图中，我可以看到： https: //www.dropbox.com/s/zvny7qwhl7hbwyw/Screenshot%202018-07-08%2010.05.33.png ?dl=0

日志显示： E Out of memory: Kill process 1180 (java) score 1100 or sacrifice child E Killed process 1180 (java) total-vm:4838044kB, anon-rss:383132kB, file-rss:0kB 更多信息请参见：https ://pastebin.com/raw/MftBwUxs

我怎样才能调试发生了什么？

我有一个程序，它在 pubSub 中创建一个主题，并向该主题发布消息。我还有一个自动数据流作业（使用模板），它将这些消息保存到我的 BigQuery 表中。现在我打算用 python 管道替换基于模板的作业，其中我的要求是从 PubSub 读取数据，应用转换并将数据保存到 BigQuery/发布到另一个 PubSub 主题。我开始用 python 编写脚本，并进行了大量的试验和错误来实现它，但令我沮丧的是，我无法实现它。代码如下所示：

import apache_beam as beam
from apache_beam.io import WriteToText
TOPIC_PATH = "projects/test-pipeline-253103/topics/test-pipeline-topic"
OUTPUT_PATH = "projects/test-pipeline-253103/topics/topic-repub"

def run():
    o = beam.options.pipeline_options.PipelineOptions()
    p = beam.Pipeline(options=o)

    print("I reached here")
    # # Read from PubSub into a PCollection.
    data = (
        p
        | "Read From Pub/Sub" >> beam.io.ReadFromPubSub(topic=TOPIC_PATH)
    )
    data | beam.io.WriteToPubSub(topic=OUTPUT_PATH)
    print("Lines: ", data)
run()

如果我能尽早获得一些帮助，我将非常感激。注意：我在谷歌云上设置了我的项目，并且我的脚本在本地运行。

我正在使用 Python SDK 2.15.0 构建数据流管道。在此管道中，我需要在管道的多个阶段将附加数据加入到每个元素。

所有这些附加数据都是从 Google Cloud Storage 上的 avro 文件中读取的（Dataflow 和 GCS 存储桶使用的同一区域），使用 map 函数将其组织为键值元组，然后使用 pvalue.AsDict( 作为侧输入传递到 DoFn ）。侧面输入数据在管道执行期间不会改变。

第一次连接（侧面输入大小 ~ 1 MB）进行得非常顺利。然而，第二次连接确实表现不佳。它的 sideinput 大小约为 50 MB。

数据流执行图清楚地显示了导致性能不佳的原因：我的 ParDo 步骤消耗的大约 90% 的时间都花在了读取侧面输入上。即使我只使用四个工作节点，从 sideinput 读取的数据量也超出了其实际大小几个数量级。

我能做些什么来防止这种情况发生吗？我是否需要以某种方式配置工作缓存大小？在我的 DoFn 的设置方法中准备附加数据而不是将其作为 sideinput 传递会更好吗？

以下是我准备侧面输入的方法：

sideinput_1 = pvalue.AsDict(p | "Read side input data 1" >> beam.io.ReadFromAvro("gs:/bucket/small_file.avro",0,False,True) \
                              | "Prepare sideinput 1" >> beam.Map(lambda x: (x["KEY"],x["VALUE"])))

# Preparing data for later join
sideinput_2 = pvalue.AsDict(p | "Read side input data 2" >> beam.io.ReadFromAvro("gs://bucket/bigger_file.avro",0,False,True) \ …

我正在 Google Dataflow 上使用 Apache Beam，并通过 lambda 函数调用函数情感，但收到错误消息：函数名称未定义。

output_tweets = (lines
                     | 'decode' >> beam.Map(lambda x: x.decode('utf-8'))
                     | 'assign window key' >> beam.WindowInto(window.FixedWindows(10))
                     | 'batch into n batches' >> BatchElements(min_batch_size=49, max_batch_size=50)
                     | 'sentiment analysis' >> beam.FlatMap(lambda x: sentiment(x))
                     )

这是我的 Apache Beam 调用，在最后一行中提到了函数情绪，这给我带来了问题。

函数代码如下（我认为这不重要）：

def sentiment(messages):
    if not isinstance(messages, list):
        messages = [messages]

    instances = list(map(lambda message: json.loads(message), messages))
    lservice = discovery.build('language', 'v1beta1', developerKey = APIKEY)
    for instance in instances['text']:
        response = lservice.documents().analyzeSentiment(
            body ={
                'document': {
                    'type': 'PLAIN_TEXT',
                    'content': …

我正在尝试使用该apache_beam.io.fileio模块来读取文件lines.txt并将其合并到我的管道中。

lines.txt有以下内容：

line1
line2
line3

当我运行以下管道代码时：

with beam.Pipeline(options=pipeline_options) as p:

     lines = (
         p
         | beam.io.fileio.MatchFiles(file_pattern="lines.txt")
         | beam.io.fileio.ReadMatches()
     )
     # print file contents to screen
     lines | 'print to screen' >> beam.Map(print)

我得到以下输出：

<apache_beam.io.fileio.ReadableFile object at 0x000001A8C6C55F08>

我期望

line1
line2
line3

我怎样才能达到我预期的结果？

窗格和窗户有什么区别？传入的元素被分组到窗口中。那么窗格包含什么呢？

我从 Beam 文档中获取了以下代码

.of(new DoFn<String, String>() {
     public void processElement(@Element String word, PaneInfo paneInfo) {
  }})

每个元素都属于一个窗格吗？还是多个窗格？需要一个简单的类比来理解窗格和窗口

使用BigQueryIO时如何选择这2个写入函数。

找到了 官方文档 这里提到writeTableRows()不推荐，但不明白为什么。

1. 我可以用来gson序列化我的 Java POJO并直接TableRow写入writeTableRows()
2. TableRow如果我使用，我需要手动设置对象write()

问题：

1. write()喜欢和选择的理由是什么
2. 这两个选项有很大的性能差异吗？

谢谢

我使用Google Dataflow(apache-beam)设置了一个小测试.该实验的用例是获取(csv)文件并将选定列写入(txt)文件.

实验代码如下:

from __future__ import absolute_import

import argparse
import logging
import re

import apache_beam as beam
from apache_beam.io import ReadFromText
from apache_beam.io import WriteToText
from apache_beam.metrics import Metrics
from apache_beam.metrics.metric import MetricsFilter
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions

class EmitColDoFn(beam.DoFn):
    first = True
    header = ""
    def __init__(self, i):
        super(EmitColDoFn, self).__init__()
        self.line_count =  Metrics.counter(self.__class__, 'lines')
        self.i = i

    def process(self, element):
        if self.first:
            self.header = element
            self.first = False
        else:
            self.line_count.inc()
            cols = re.split(',', element)
            return (cols[self.i],) …

我正在对数据流批处理负载进行性能基准测试,发现与Bigquery命令行工具上的相同负载相比,负载太慢了.

文件大小约为20 MB,有数百万条记录.我尝试了不同的机器类型,并且n1-highmem-4在加载目标BQ表时加载时间为8分钟时获得了最佳的负载性能.

通过在命令行实用程序上运行BQ命令来应用相同的表加载时,处理和加载相同数量的数据几乎不需要2分钟.有关使用Dataflow作业的负载性能不佳的任何见解？如何提高性能使其与BQ命令行实用程序相媲美？