问题标签 [google-cloud-dataflow]

我已经浏览了 Java SDK 文档，但要么遗漏了一些明显的东西，要么没有看到。布尔值的合适编码器是什么？这似乎足够基本，我假设前者是这种情况。

我正在尝试将 google memcache 与数据流一起使用。我本质上想将数据转换为内存缓存。是否可以在数据流中使用 google memcache api？

我收到以下错误：java.util.concurrent.ExecutionException： com.google.apphosting.api.ApiProxy$CallNotFoundException: The API package 'memcache' or call 'Set()' was not found. com.google.appengine.api.utils.FutureWrapper.setExceptionResult(FutureWrapper.java:65)

这是代码行：

我正在尝试将一些数据放入数据流中，但数据不在云存储中 - 它是一个 rss 提要，我通常每 x 小时检查一次。有没有办法直接使用 SDK 来做到这一点，或者我必须先以其他方式将文件放到云存储中。

提前致谢。

It turns out that I am not able to run any transformations in my pipeline presumably because Cloud Dataflow currently runs only Java 7. All my pre-existing code that runs in my pipeline relies on Java 8 (can only be built with the Java 8 compiler). Is there any way around it? I saw the Java 7 restriction clearly stated here: https://cloud.google.com/dataflow/faq, but I'm hoping there's some hacky way I could get around it.

如何获得对运行 Dataflow 工作器的 Docker 容器的 shell 访问权限？

正如这个问题的评论中提到的，我们可以设置teardownPolicy=TEARDOWN_NEVER在管道完成后保留一个工作虚拟机。这允许我们通过 SSH 连接到主机 VM，但它不会将我们插入到运行工作任务的 Docker 容器中。如何获得对该容器的 shell 访问权限？

上下文：我想在我的工作人员内使用 Docker 容器，但在工作人员容器内docker似乎没有可用的二进制文件。我试图在容器中安装 Docker，但是我遇到了错误，并且为了测试下一个命令而必须启动一个新的管道很慢。

我正在构建一个示例数据流管道，主要基于 https://cloud.google.com/dataflow/java-sdk/combine上的代码

但是当我运行我的代码时，我遇到了以下异常：

线程“主”java.lang.IllegalArgumentException 中的异常：无法在 com.google.cloud.dataflow.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils) 处序列化 com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner$TestCombineDoFn@139982de .java:51) 在 com.google.cloud.dataflow.sdk.util.SerializableUtils.ensureSerializable(SerializableUtils.java:81) 在 com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner$Evaluator.ensureSerializable(DirectPipelineRunner.java :784) 在 com.google.cloud.dataflow.sdk.transforms.ParDo.evaluateHelper(ParDo.java:1025) 在 com.google.cloud.dataflow.sdk.transforms.ParDo.evaluateSingleHelper(ParDo.java:963) 在com.google.cloud.dataflow.sdk.transforms.ParDo.access$000(ParDo.java:441) 在 com.google.cloud.dataflow.sdk.transforms.ParDo$1。com.google.cloud.dataflow.sdk.transforms.ParDo$1.evaluate(ParDo.java:946) 上的评估（ParDo.java:951） com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner$Evaluator.visitTransform (DirectPipelineRunner.java:611) 在 com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:200) 在 com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java :196) 在 com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:196) 在 com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:196) 在com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:196) 在 com.google.cloud.dataflow.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:109) 在 com.google。云。dataflow.sdk.Pipeline.traverseTopologically(Pipeline.java:204) 在 com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner$Evaluator.run(DirectPipelineRunner.java:584) 在 com.google.cloud.dataflow.sdk。 com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner.run(DirectPipelineRunner.java:70) 在 com.google.cloud.dataflow.sdk.Pipeline.run( Pipeline.java:145) 在 com.google.cloud.dataflow.examples.CalcMeanExample.main(CalcMeanExample.java:50) 原因：java.io.NotSerializableException: org.apache.avro.io.DecoderFactory 在 java.io。 java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) 中的 ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)java:1509) 在 java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) 在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) 在 java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) 在java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) 在 java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) 在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) 在 java.io.ObjectOutputStream .writeObject(ObjectOutputStream.java:348) 在 com.google.cloud.dataflow.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:47) ... 20 更多writeObject0(ObjectOutputStream.java:1178) 在 java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) 在 java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) 在 java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java :1432) 在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) 在 java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) 在 com.google.cloud.dataflow.sdk.util.SerializableUtils.serializeToByteArray( SerializableUtils.java:47) ... 20 更多writeObject0(ObjectOutputStream.java:1178) 在 java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) 在 java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) 在 java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java :1432) 在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) 在 java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) 在 com.google.cloud.dataflow.sdk.util.SerializableUtils.serializeToByteArray( SerializableUtils.java:47) ... 20 更多1432) 在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) 在 com.google.cloud.dataflow.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils) .java:47) ... 20 更多1432) 在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) 在 com.google.cloud.dataflow.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils) .java:47) ... 20 更多

我的代码如下：

{

}

有什么想法会导致这种情况吗？

我正在尝试使用谷歌云数据流从 GCS 读取数据并加载到 BigQuery 表，但是 GCS 中的文件是压缩的（gzip），是否有任何类可用于从压缩/gzip 压缩文件中读取数据？

我使用数据流服务运行我的工作以从 GCS 读取数据并写入 BQ 表，工作成功运行，我想知道如何或在哪里可以找到谷歌为我的工作设置了多少虚拟机的信息？

谢谢你。

我可以在开发人员控制台中看到项目“ Dataflow: Readonly Artifacts (DO NOT DELETE) ”是否正确？自从我上个月获得 CDF 的 alpha 访问权限以来，它就一直可见。

我还注意到，甚至在我们开始测试和运行作业之前（实际上只是几个小时前！），我就已经对这个项目负责了。我的理解是 CDF 在 alpha 版中是免费的，但是您必须为使用的任何服务付费，例如 BigQuery、GCS 等。但是，我希望这些费用会出现在我的其他项目中，因为我在实际创建时指定了该项目名称/ID管道。

我有一个大数据文件 (1 TB) 的数据要导入 BigQuery。每行包含一个键。在导入数据并创建要导出到 BigQuery 的 PCollection 时，我想确保不会基于此键值导入重复记录。使用 Dataflow 在我的 Java 程序中执行此操作的最有效方法是什么？

谢谢