是否可以设置BigQuery JobID或在批处理管道运行时获取它.
我知道使用BigQuery API是可能的,但如果我使用Apache Beam的BigQueryIO,它是否可能？我需要在写完BigQuery后发送确认信息表明加载完成了.
在Apache的梁编程指南包含以下规则：
3.2.2。不变性
PCollection是不可变的。创建后，您将无法添加，删除或更改单个元素。Beam变换可以处理PCollection的每个元素并生成新的管道数据（作为新的PCollection），但是它不会消耗或修改原始的输入集合。
这是否意味着我不能，一定不能或不应该在自定义转换中修改单个元素？具体来说，我正在使用python SDK并考虑将dict {key: "data"}作为输入，进行一些处理并添加更多字段的转换情况{other_key: "some more data"}。我对上述规则3.2.2的解释是，我应该这样
def process(self,element):
import copy
output = copy.deepcopy(element)
output[other_key] = some_data
yield output
但我想知道这是否有点过大。
使用TestPipeline，我发现如果我在process()方法中对其进行操作，则输入集合的元素也会被修改（除非这些元素是基本类型，例如int，float，bool ...）。
变异元素被认为是绝对不行的，还是仅需谨慎的一种做法？
我正在使用Apache Beam for Java,并且正在使用Cloud DLP API和Cloud Dataflow.作业开始,但在运行时出错.
我认为将DataLink上运行的gRPC库版本与DLP API的客户端库相结合是一个问题,但我不知道要指定哪个版本.
<dependency>
<groupId>com.google.cloud</groupId>
<artifactId>google-cloud-dlp</artifactId>
<version>0.33.0-beta</version>
</dependency>
<dependency>
<groupId>com.google.api</groupId>
<artifactId>gax</artifactId>
<version>1.16.0</version>
</dependency>
<dependency>
<groupId>com.google.api</groupId>
<artifactId>gax-grpc</artifactId>
<version>0.20.0</version>
</dependency>
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>3.2.0</version>
</dependency>
java.lang.RuntimeException: java.lang.NoClassDefFoundError: com/google/api/gax/grpc/ProtoOperationTransformers$ResponseTransformer
at org.sinmetal.mlapi.DataLossPreventionFn.processElement(DataLossPreventionFn.java:52)
Caused by: java.lang.NoClassDefFoundError: com/google/api/gax/grpc/ProtoOperationTransformers$ResponseTransformer
at com.google.cloud.dlp.v2beta1.DlpServiceSettings$Builder.initDefaults(DlpServiceSettings.java:425)
at com.google.cloud.dlp.v2beta1.DlpServiceSettings$Builder.<init>(DlpServiceSettings.java:363)
at com.google.cloud.dlp.v2beta1.DlpServiceSettings$Builder.createDefault(DlpServiceSettings.java:367)
at com.google.cloud.dlp.v2beta1.DlpServiceSettings$Builder.access$000(DlpServiceSettings.java:264)
at com.google.cloud.dlp.v2beta1.DlpServiceSettings.newBuilder(DlpServiceSettings.java:233)
at com.google.cloud.dlp.v2beta1.DlpServiceClient.create(DlpServiceClient.java:149)
at org.sinmetal.mlapi.DataLossPreventionFn.processElement(DataLossPreventionFn.java:26)
at org.sinmetal.mlapi.DataLossPreventionFn$DoFnInvoker.invokeProcessElement(Unknown Source)
at org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:177)
at org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:141)
at com.google.cloud.dataflow.worker.SimpleParDoFn.processElement(SimpleParDoFn.java:324)
at com.google.cloud.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:48)
at com.google.cloud.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:52)
at com.google.cloud.dataflow.worker.SimpleParDoFn$1.output(SimpleParDoFn.java:272)
at org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue(SimpleDoFnRunner.java:211)
at org.apache.beam.runners.core.SimpleDoFnRunner.access$700(SimpleDoFnRunner.java:66)
at org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:436)
at org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:424)
at org.apache.beam.sdk.io.gcp.bigquery.PassThroughThenCleanup$IdentityFn.processElement(PassThroughThenCleanup.java:83)
at …我正在尝试使用Apache Beam在Dataflow中编写作业.此作业需要获取输入并将其转换为我的自定义对象.此对象表示内存测试,其中包含固定属性,如timestamp,name ...以及具有其属性的分区列表
public class TestResult {
String testName;
String testId;
String testStatus;
String testResult;
List<Partition> testPartitions;
}
public class Partition {
String testId;
String filesystem;
String mountedOn;
String usePercentage;
String available;
String size;
String used;
}
我的最后一个转换,获取此TestResult对象并将其转换为表行.
static class TestResultToRowConverter extends DoFn<TestResult, TableRow> {
/**
* In this example, put the whole string into single BigQuery field.
*/
@ProcessElement
public void processElement(ProcessContext c) {
System.out.println("setting TestResult-> TestResult:" + c.element());
c.output(new TableRow().set("testName", c.element().testName).set("testId", c.element().testId).set("testStatus", c.element().testStatus).set("testResult", c.element().testResult).set("memoryTestData", "example data test"));
for …如何在 BEAM SQL 中的 GroupByKey 之前包含 Window.into 或 Window.triggering 转换？
我有以下 2 个表：
表
CREATE TABLE table1(
field1 varchar
,field2 varchar
)
第二表
CREATE TABLE table2(
field1 varchar
,field3 varchar
)
我正在将结果写在第三个表中
CREATE TABLE table3(
field1 varchar
,field3 varchar
)
前 2 个表正在从 kafka 流中读取数据，我正在对这些表进行连接并将数据插入到第三个表中，使用以下查询。前 2 个表是无界/无界的
INSERT INTO table3
(field1,
field3)
SELECT a.field1,
b.field3
FROM table1 a
JOIN table2 b
ON a.field1 = b.field1
我收到以下错误：
引起：java.lang.IllegalStateException：GroupByKey 不能应用于没有触发器的 GlobalWindow 中的无界 PCollection。在 GroupByKey 之前使用 Window.into 或 Window.triggering 转换。在 org.apache.beam.sdk.transforms.GroupByKey.applicableTo(GroupByKey.java:173) 在 org.apache.beam.sdk.transforms.GroupByKey.expand(GroupByKey.java:204) …
我正在使用Apache Beam 2.6从单个Kafka主题中读取并将输出写入Google Cloud Storage(GCS).现在我想改变管道,以便它正在读取多个主题并将其写出来gs://bucket/topic/...
在阅读我TextIO在管道的最后一步中使用的单个主题时:
TextIO.write()
.to(
new DateNamedFiles(
String.format("gs://bucket/data%s/", suffix), currentMillisString))
.withWindowedWrites()
.withTempDirectory(
FileBasedSink.convertToFileResourceIfPossible(
String.format("gs://bucket/tmp%s/%s/", suffix, currentMillisString)))
.withNumShards(1));
这是一个类似的问题,我试图改编的代码.
FileIO.<EventType, Event>writeDynamic()
.by(
new SerializableFunction<Event, EventType>() {
@Override
public EventType apply(Event input) {
return EventType.TRANSFER; // should return real type here, just a dummy
}
})
.via(
Contextful.fn(
new SerializableFunction<Event, String>() {
@Override
public String apply(Event input) {
return "Dummy"; // should return the Event converted to a String
}
}),
TextIO.sink())
.to(DynamicFileDestinations.constant(new DateNamedFiles("gs://bucket/tmp%s/%s/",
currentMillisString), …我无法按照以下说明使用 wordcount 示例创建自定义 Google Cloud Dataflow 模板：https : //cloud.google.com/dataflow/docs/guides/templates/creating-templates
我收到与无法访问 RuntimeValueProvider 相关的错误。我究竟做错了什么？
我的主要功能wordcount.py：
"""A word-counting workflow."""
from __future__ import absolute_import
import argparse
import logging
import re
from past.builtins import unicode
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, GoogleCloudOptions
from apache_beam.options.pipeline_options import SetupOptions
class WordExtractingDoFn(beam.DoFn):
"""Parse each line of input text into words."""
def __init__(self):
self.words_counter = Metrics.counter(self.__class__, 'words')
self.word_lengths_counter = …我试图找出 Map 和 ParDo 之间的性能差异，但我无法以某种方式运行 ParDo 方法
我已经尝试寻找一些尝试解决问题的资源，但我没有找到
ParDo 方法（这不起作用）：
class ci(beam.DoFn):
def compute_interest(self,data_item):
cust_id, cust_data = data_item
if(cust_data['basic'][0]['acc_opened_date']=='2010-10-10'):
new_data = {}
new_data['new_balance'] = (cust_data['account'][0]['cur_bal'] * cust_data['account'][0]['roi']) / 100
new_data.update(cust_data['account'][0])
new_data.update(cust_data['basic'][0])
del new_data['cur_bal']
return new_data
地图方法（这有效）：
def compute_interest(data_item):
cust_id, cust_data = data_item
if(cust_data['basic'][0]['acc_opened_date']=='2010-10-10'):
new_data = {}
new_data['new_balance'] = (cust_data['account'][0]['cur_bal'] * cust_data['account'][0]['roi']) / 100
new_data.update(cust_data['account'][0])
new_data.update(cust_data['basic'][0])
del new_data['cur_bal']
return new_data
错误：
引发 NotImplementedError RuntimeError: NotImplementedError [运行 'PIPELINE NAME']
我在Java上使用Apache Beam。我正在尝试使用本地模式在预先部署的Spark env上使用SparkRunner读取csv文件并将其写入拼花格式。DirectRunner一切正常，但是SparkRunner无法正常工作。我正在使用Maven Shade插件构建胖子。
代码如下：
Java：
public class ImportCSVToParquet{
-- ommitted
File csv = new File(filePath);
PCollection<String> vals = pipeline.apply(TextIO.read().from(filePath));
String parquetFilename = csv.getName().replaceFirst("csv", "parquet");
String outputLocation = FolderConventions.getRawFilePath(confETL.getHdfsRoot(), parquetFilename);
PCollection<GenericRecord> processed = vals.apply(ParDo.of(new ProcessFiles.GenericRecordFromCsvFn()))
.setCoder(AvroCoder.of(new Config().getTransactionSchema()));
LOG.info("Processed file will be written to: " + outputLocation);
processed.apply(FileIO.<GenericRecord>write().via(ParquetIO.sink(conf.getTransactionSchema())).to(outputLocation));
pipeline.run().waitUntilFinish();
}
POM依赖项：
<dependencies>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-core</artifactId>
<version>2.14.0</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-direct-java</artifactId>
<version>2.14.0</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-spark</artifactId>
<version>2.14.0</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-parquet</artifactId>
<version>2.14.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.2.3</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.2.3</version> …我已经使用Beam一段时间了，我想知道编写高效且优化的Beam管道的关键概念是什么。
我有一些Spark背景知识，并且我知道我们可能更喜欢使用reduceByKey而不是groupByKey以避免混洗并优化网络流量。
Beam也一样吗？
我将不胜感激一些技巧或材料/最佳实践。