Guillem Xercavins通过此链接编写了一个用于计算最小值和最大值的自定义类。
class MinMaxFn(beam.CombineFn):
# initialize min and max values (I assumed int type)
def create_accumulator(self):
return (sys.maxint, 0)
# update if current value is a new min or max
def add_input(self, min_max, input):
(current_min, current_max) = min_max
return min(current_min, input), max(current_max, input)
def merge_accumulators(self, accumulators):
return accumulators
def extract_output(self, min_max):
return min_max
我还需要计算平均值，我发现示例代码如下：
class MeanCombineFn(beam.CombineFn):
def create_accumulator(self):
"""Create a "local" accumulator to track sum and count."""
return (0, 0)
def add_input(self, (sum_, count), input):
"""Process the incoming …我正在尝试通过 python 中的 apache beam 读取 JSON 文件，并对其应用一些数据质量规则。目前我正在使用 beam.io.ReadFromText 读取每个 json 行并使用一些函数来修改数据。读取 JSON 数据并修改它们的更好方法是什么？
(p
| 'Getdata' >> beam.io.ReadFromText(input)
| 'filter_name' >> beam.FlatMap(lambda line: dq_name(line))
| 'filter_phone' >> beam.FlatMap(lambda line: dq_phone(line))
| 'filter_zip' >> beam.FlatMap(lambda line: dq_zip(line))
| 'filter_address' >> beam.FlatMap(lambda line: dq_city(line))
| 'filter_website' >> beam.FlatMap(lambda line: dq_website(line))
| 'write' >> beam.io.WriteToText(output_prefix) )
注意：我对此还很陌生，如果我当前的方法看起来太粗制滥造，我很抱歉。
python google-cloud-platform google-cloud-dataflow apache-beam
我正在 Apache Beam 中编写一个数据管道，它从 Pub/Sub 读取数据，将消息反序列化为 JSONObjects 并将它们传递到其他一些管道阶段。问题是，当我尝试提交代码时，出现以下错误：
执行 Java 类时发生异常。无法返回用于转换为 JSON 并混淆 PII 数据/ParMultiDo(JSONifyAndObfuscate).output [PCollection] 的默认编码器。更正以下根本原因之一： [错误] 未手动指定编码器；您可以使用 .setCoder() 来执行此操作。[错误] 从 CoderRegistry 推断编码器失败：无法为 org.json.JSONObject 提供编码器。[错误] 使用注册的 CoderProvider 构建 Coder 失败。[错误] 请参阅抑制的异常以了解详细的失败信息。[错误] 使用生成 PTransform 的默认输出编码器失败：调用了 PTransform.getOutputCoder。
基本上，错误表明 Beam 无法找到 org.json.JSONObject 对象的编码器。我不知道在哪里可以获得这样的编码器或如何构建一个。有任何想法吗？
谢谢！
我正在尝试在云数据流上运行 apache-beam 管道。原始函数已部署为云函数，该函数应该创建一个读取文本文件并插入大查询的数据流作业。但它无法在 Dataflow 上运行。下面给出了功能和错误。
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import GoogleCloudOptions
from apache_beam.options.pipeline_options import StandardOptions
import apache_beam as beam
class Split(beam.DoFn):
def process(self, element):
element = element.split(',')
return [{
'field_1': element[0],
'field_2': element[1],
'field_3': element[2]}]
def main(data, context):
options = PipelineOptions()
google_cloud_options = options.view_as(GoogleCloudOptions)
google_cloud_options.project = my-project
google_cloud_options.job_name = job_name
google_cloud_options.staging_location = staging_location
google_cloud_options.temp_location = temp_location
google_cloud_options.service_account_email = service_account_email
options.view_as(StandardOptions).runner = 'DataflowRunner'
p = beam.Pipeline(options=options)
with p:
(
p
| 'ReadData' >> beam.io.ReadFromText(gs://source_file_location)
| 'ParseCSV' >> beam.ParDo(Split())
| …我注意到java apache beam有类groupby.sortbytimestamp python是否已实现该功能？如果不是在窗口中对元素进行排序的方法是什么？我想我可以在DoFn中对整个窗口进行排序,但我想知道是否有更好的方法.
我正在尝试使用apache-beam创建一个流管道,从google pub/sub读取句子并将这些单词写入Bigquery表.
我正在使用0.6.0apache-beam版本.
按照这些例子,我做了这个:
public class StreamingWordExtract {
/**
* A DoFn that tokenizes lines of text into individual words.
*/
static class ExtractWords extends DoFn<String, String> {
@ProcessElement
public void processElement(ProcessContext c) {
String[] words = ((String) c.element()).split("[^a-zA-Z']+");
for (String word : words) {
if (!word.isEmpty()) {
c.output(word);
}
}
}
}
/**
* A DoFn that uppercases a word.
*/
static class Uppercase extends DoFn<String, String> {
@ProcessElement
public void processElement(ProcessContext c) {
c.output(c.element().toUpperCase());
}
} …我尝试使用以下内容
TextIO.Read.from("gs://xyz.abc/xxx_{2017-06-06,2017-06-06}.csv")
我得到的那种模式不起作用
java.lang.IllegalStateException: Unable to find any files matching StaticValueProvider{value=gs://xyz.abc/xxx_{2017-06-06,2017-06-06}.csv}
即使这两个文件确实存在.我尝试使用类似表达式的本地文件
TextIO.Read.from("somefolder/xxx_{2017-06-06,2017-06-06}.csv")
这确实很好用.
我原以为GCS中的文件会有各种各样的支持,但不是.这是为什么？在那里完成我正在寻找的东西？
这个问题是一个跟进到这一个.我正在尝试使用apache beam从google spanner表中读取数据(然后进行一些数据处理).我使用java SDK编写了以下最小示例:
package com.google.cloud.dataflow.examples;
import java.io.IOException;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.values.PCollection;
import com.google.cloud.spanner.Struct;
public class backup {
public static void main(String[] args) throws IOException {
PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
PCollection<Struct> rows = p.apply(
SpannerIO.read()
.withInstanceId("my_instance")
.withDatabaseId("my_db")
.withQuery("SELECT t.table_name FROM information_schema.tables AS t")
);
PipelineResult result = p.run();
try {
result.waitUntilFinish();
} catch (Exception exc) {
result.cancel();
}
}
}
当我尝试使用DirectRunner执行代码时,我收到以下错误消息:
org.apache.beam.runners.direct.repackaged.com.google.common.util.concurrent.UncheckedExecutionException:
org.apache.beam.sdk.util.UserCodeException:java.lang.NoClassDefFoundError:无法初始化类com.google.cloud.spanner.spi.v1.SpannerErrorInterceptor
[...]引起:org.apache.beam.sdk.util.UserCodeException:java.lang.NoClassDefFoundError:无法初始化类com.google.cloud.spanner.spi.v1.SpannerErrorInterceptor
[...]引起:java.lang.NoClassDefFoundError:无法初始化类com.google.cloud.spanner.spi.v1.SpannerErrorInterceptor
或者,使用DataflowRunner: …
如何在Apache Beam中实现Pandas？我无法在多个列上执行左联接，并且Pcollections不支持sql查询。甚至Apache Beam文档也没有正确地构建框架。我检查了一下，但是在Apache Beam中找不到任何熊猫实现。谁能将我定向到所需的链接？
我有一个PCollection，我想使用ParDo从中筛选出一些元素。
在这里可以找到一个例子吗？