标签: apache-beam

我注意到在 java sdk 中，有一个函数可以让您编写 csv 文件的标题。 https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/io/TextIO.Write.html#withHeader-java.lang.String-

此功能是否反映在 python skd 上？

我正在使用带有Python SDK的Google Cloud Dataflow.

我想要 :

• 从主PCollection中获取唯一日期列表
• 循环遍历该列表中的日期以创建过滤的PCollections(每个都具有唯一的日期),并将每个过滤的PCollection写入BigQuery中时间分区表中的分区.

我怎样才能获得该列表？在下面的combine变换之后,我创建了一个ListPCollectionView对象但是我无法迭代该对象:

class ToUniqueList(beam.CombineFn):

    def create_accumulator(self):
        return []

    def add_input(self, accumulator, element):
        if element not in accumulator:
            accumulator.append(element)
        return accumulator

    def merge_accumulators(self, accumulators):
        return list(set(accumulators))

    def extract_output(self, accumulator):
        return accumulator


def get_list_of_dates(pcoll):

    return (pcoll
            | 'get the list of dates' >> beam.CombineGlobally(ToUniqueList()))

我做错了吗？最好的方法是什么？

谢谢.

可以通过以下方式在Data Storage上读取未存储的JSON文件:

p.apply("read logfiles", TextIO.Read.from("gs://bucket/*").withCoder(TableRowJsonCoder.of()));

如果我只想用BigQuery编写那些带有最小过滤的日志,我可以通过使用像这样的DoFn来实现:

private static class Formatter extends DoFn<TableRow,TableRow> {

        @Override
        public void processElement(ProcessContext c) throws Exception {

            // .clone() since input is immutable
            TableRow output = c.element().clone();

            // remove misleading timestamp field
            output.remove("@timestamp");

            // set timestamp field by using the element's timestamp
            output.set("timestamp", c.timestamp().toString());

            c.output(output);
        }
    }
}

但是,我不知道如何以这种方式访问​​JSON文件中的嵌套字段.

1. 如果TableRow包含一个RECORDnamed r,是否可以访问其键/值而无需进一步序列化/反序列化？
2. 如果我需要序列化/反序列化自己与Jackson库,它让使用标准更有意义Coder的TextIO.Read替代TableRowJsonCoder,从而获得一些我失去这样的演出回来的？

编辑

文件是换行符的新行,看起来像这样:

{"@timestamp":"2015-x", "message":"bla", "r":{"analyzed":"blub", "query": {"where":"9999"}}}
{"@timestamp":"2015-x", "message":"blub", "r":{"analyzed":"bla", "query": {"where":"1111"}}}

我正在使用Python Beam SDK 0.6.0。我想将输出写入Google Cloud Storage中的JSON。做这个的最好方式是什么？

我可以WriteToText在Text IO接收器中使用，但是然后我必须分别格式化每一行，对吗？如何将结果保存到包含对象列表的有效JSON文件中？

I am trying to use apache beam's google datastore api to ReadFromDatastore

p = beam.Pipeline(options=options)
(p
 | 'Read from Datastore' >> ReadFromDatastore(gcloud_options.project, query)
 | 'reformat'            >> beam.Map(reformat)
 | 'Write To Datastore'  >> WriteToDatastore(gcloud_options.project))

The object that gets passed to my reformat function is type

google.cloud.proto.datastore.v1.entity_pb2.Entity

It is in protobuf format which is hard to modify or read.

I think I can convert a entity_pb2.Entity to a dict with

entity= dict(google.cloud.datastore.helpers._property_tuples(entity_pb))

But for some reason trying to import the following two …

我目前正在尝试运行数据流（Apache Beam，Python SDK）任务，以将大于100GB的Tweet文件导入BigQuery，但正在 Error: Message: Too many sources provided: 15285. Limit is 10000.

该任务将使用tweet（JSON），提取5个相关字段，并通过一些转换对它们进行一些转换/消毒，然后将这些值写入BigQuery，以用于进一步处理。

BigQuery有Cloud Dataflow-来源过多，但这似乎是由于输入文件过多而引起的，而我只有一个输入文件，因此似乎无关紧要。另外，这里提到的解决方案是很神秘的，我不确定是否/如何将它们应用于我的问题。

我的猜测是，BigQuery在持久存储之前为每一行或其他内容写入临时文件，这就是“太多源”的含义吗？

我怎样才能解决这个问题？

[编辑]

码：

import argparse
import json
import logging

import apache_beam as beam

class JsonCoder(object):
    """A JSON coder interpreting each line as a JSON string."""

    def encode(self, x):
        return json.dumps(x)

    def decode(self, x):
        return json.loads(x)

def filter_by_nonempty_county(record):
    if 'county_fips' in record and record['county_fips'] is not None:
        yield record

def run(argv=None):

    parser = argparse.ArgumentParser()
    parser.add_argument('--input',
                        default='...',
                        help=('Input twitter json …

我正在尝试从 DoFn 方法获得两个输出，遵循Apache Beam 编程指南的示例

基本上在示例中，您传递一个 TupleTag，然后指定在哪里进行输出，这对我有用，问题是我在 ParDo 中调用了一个外部方法，但不知道如何传递这个 TupleTag，这是我的代码：

PCollectionTuple processedData = pubEv
  .apply("Processing", ParDo.of(new HandleEv())
      .withOutputTags(mainData, TupleTagList.of(failedData)));

HandleEv 方法：

static class HandleEv extends DoFn<String, String> {
    @ProcessElement
    public void processElement(ProcessContext c) throws Exception {
      c.output("test")
      c.output(failedData,"failed")
    }
}

我得到的错误是cannot find symbol由于 failedData 不能从 HandleEv 访问，我试图在课程开始时声明 failedData 但也不起作用。

非常感谢

我有一个P1包含 ID 字段的PCollection 。我想从 PCollection 中获取完整的 ID 列作为列表，并将此值传递给 BigQuery 查询以过滤一个 BigQuery 表。

这样做的最快和最优化的方法是什么？

我是 Dataflow 和 BigData 的新手。任何人都可以对此提供一些提示吗？

谢谢！

我是Apache Beam的新手，我在其中尝试编写管道以从Google BigQuery提取数据，然后使用Python将数据以CSV格式写入GCS。

使用，beam.io.read(beam.io.BigQuerySource())我能够从BigQuery读取数据，但不确定如何将其以CSV格式写入GCS。

是否有实现相同功能的自定义功能，能否请您帮我吗？

import logging

import apache_beam as beam


PROJECT='project_id'
BUCKET='project_bucket'


def run():
    argv = [
        '--project={0}'.format(PROJECT),
        '--job_name=readwritebq',
        '--save_main_session',
        '--staging_location=gs://{0}/staging/'.format(BUCKET),
        '--temp_location=gs://{0}/staging/'.format(BUCKET),
        '--runner=DataflowRunner'
]

with beam.Pipeline(argv=argv) as p:

    # Execute the SQL in big query and store the result data set into given Destination big query table.
    BQ_SQL_TO_TABLE = p | 'read_bq_view' >> beam.io.Read(
        beam.io.BigQuerySource(query =  'Select * from `dataset.table`', use_standard_sql=True))
    # Extract data from Bigquery to GCS in CSV format.
    # This is where I need …

我有一个从pubsub获得的Object的PCollection，可以这样说：

 PCollection<Student> pStudent ;

在学生属性中，有一个属性，比如说studentID；并且我想使用此学生ID从BigQuery读取属性（class_code），并将我从BQ获取的class_code设置为PCollcetion中的Student Object

有谁知道如何实现这一目标？我知道在Beam中有一个，BigQueryIO但是如果我要在BQ中执行的查询字符串条件来自PCollection中的学生对象（studentID），那么我该怎么办？如何从BigQuery的结果中将值设置为PCollection ？