【发布时间】：2015-12-28 14:19:43
【问题描述】：
我按照这篇文章搭建了一个Cloud Dataflow Pipeline：https://cloud.google.com/bigtable/docs/dataflow-hbase
当我将其提交到 Cloud Dataflow 托管服务时，我在 Cloud Dataflow 工作人员处收到以下错误：
Uncaught exception in main thread. Exiting with status code 1.
java.lang.NoSuchMethodError: io.grpc.netty.GrpcSslContexts.forClient()Lcom/google/bigtable/repackaged/io/netty/handler/ssl/SslContextBuilder;
at com.google.cloud.bigtable.grpc.BigtableSession.createSslContext(BigtableSession.java:98)
at com.google.cloud.bigtable.grpc.BigtableSession.access$000(BigtableSession.java:82)
at com.google.cloud.bigtable.grpc.BigtableSession$1.run(BigtableSession.java:151)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
我应该如何处理这个问题？
我的 Cloud Dataflow Pipeline 源代码如下：
package mypackage
import com.google.cloud.bigtable.dataflow.CloudBigtableIO;
import com.google.cloud.bigtable.dataflow.CloudBigtableOptions;
import com.google.cloud.bigtable.dataflow.CloudBigtableTableConfiguration;
import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
import com.google.cloud.dataflow.sdk.transforms.Create;
import com.google.cloud.dataflow.sdk.transforms.DoFn;
import com.google.cloud.dataflow.sdk.transforms.ParDo;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
public class Main {
// Create a DoFn that creates a Put or Delete. MUTATION_TRANSFORM is a simplistic example.
static final DoFn<String, Mutation> MUTATION_TRANSFORM = new DoFn<String, Mutation>() {
@Override
public void processElement(DoFn<String, Mutation>.ProcessContext c) throws Exception {
c.output(new Put(c.element().getBytes()).addColumn("v".getBytes(), "v".getBytes(), "value".getBytes()));
}
};
public static void main(String[] args) {
// CloudBigtableOptions is one way to retrieve the options. It's not required to use this
// specific PipelineOptions extension; CloudBigtableOptions is there as a convenience.
CloudBigtableOptions options =
PipelineOptionsFactory.fromArgs(args).withValidation().as(CloudBigtableOptions.class);
// CloudBigtableTableConfiguration contains the project, zone, cluster and table to connect to
CloudBigtableTableConfiguration config = CloudBigtableTableConfiguration.fromCBTOptions(options);
Pipeline p = Pipeline.create(options);
// This sets up serialization for Puts and Deletes so that Dataflow can potentially move them through
// the network
CloudBigtableIO.initializeForWrite(p);
p
.apply(Create.of("Hello", "World"))
.apply(ParDo.of(MUTATION_TRANSFORM))
.apply(CloudBigtableIO.writeToTable(config));
p.run();
}
}
【问题讨论】：