Repository: incubator-beam Updated Branches: refs/heads/master c0efe568e -> ee198d4e9
[BEAM-48] Remove BigQueryIOTranslator This is no longer needed for the Cloud Dataflow service. Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/22dc6a34 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/22dc6a34 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/22dc6a34 Branch: refs/heads/master Commit: 22dc6a3481a77134b0cc7d9e82865c06b15f6228 Parents: c0efe56 Author: Dan Halperin <dhalp...@google.com> Authored: Sun May 15 17:39:57 2016 -0700 Committer: Dan Halperin <dhalp...@google.com> Committed: Mon May 16 11:05:16 2016 -0700 ---------------------------------------------------------------------- .../dataflow/DataflowPipelineTranslator.java | 5 -- .../dataflow/internal/BigQueryIOTranslator.java | 72 -------------------- 2 files changed, 77 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/22dc6a34/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java ---------------------------------------------------------------------- diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java index f58ceff..d822803 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java @@ -32,7 +32,6 @@ import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; import org.apache.beam.runners.dataflow.DataflowPipelineRunner.GroupByKeyAndSortValuesOnly; -import org.apache.beam.runners.dataflow.internal.BigQueryIOTranslator; import org.apache.beam.runners.dataflow.internal.PubsubIOTranslator; import org.apache.beam.runners.dataflow.internal.ReadTranslator; import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions; @@ -42,7 +41,6 @@ import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.Pipeline.PipelineVisitor; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.IterableCoder; -import org.apache.beam.sdk.io.BigQueryIO; import org.apache.beam.sdk.io.PubsubIO; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.options.StreamingOptions; @@ -1012,9 +1010,6 @@ public class DataflowPipelineTranslator { // IO Translation. registerTransformTranslator( - BigQueryIO.Read.Bound.class, new BigQueryIOTranslator.ReadTranslator()); - - registerTransformTranslator( PubsubIO.Read.Bound.class, new PubsubIOTranslator.ReadTranslator()); registerTransformTranslator( DataflowPipelineRunner.StreamingPubsubIOWrite.class, http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/22dc6a34/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/BigQueryIOTranslator.java ---------------------------------------------------------------------- diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/BigQueryIOTranslator.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/BigQueryIOTranslator.java deleted file mode 100755 index b344f0f..0000000 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/BigQueryIOTranslator.java +++ /dev/null @@ -1,72 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.runners.dataflow.internal; - -import org.apache.beam.runners.dataflow.DataflowPipelineTranslator; -import org.apache.beam.sdk.io.BigQueryIO; -import org.apache.beam.sdk.util.PropertyNames; - -import com.google.api.services.bigquery.model.TableReference; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * BigQuery transform support code for the Dataflow backend. - */ -public class BigQueryIOTranslator { - private static final Logger LOG = LoggerFactory.getLogger(BigQueryIOTranslator.class); - - /** - * Implements BigQueryIO Read translation for the Dataflow backend. - */ - public static class ReadTranslator - implements DataflowPipelineTranslator.TransformTranslator<BigQueryIO.Read.Bound> { - - @Override - public void translate( - BigQueryIO.Read.Bound transform, DataflowPipelineTranslator.TranslationContext context) { - // Actual translation. - context.addStep(transform, "ParallelRead"); - context.addInput(PropertyNames.FORMAT, "bigquery"); - context.addInput(PropertyNames.BIGQUERY_EXPORT_FORMAT, "FORMAT_AVRO"); - - if (transform.getQuery() != null) { - context.addInput(PropertyNames.BIGQUERY_QUERY, transform.getQuery()); - context.addInput(PropertyNames.BIGQUERY_FLATTEN_RESULTS, transform.getFlattenResults()); - } else { - TableReference table = transform.getTable(); - if (table.getProjectId() == null) { - // If user does not specify a project we assume the table to be located in the project - // that owns the Dataflow job. - String projectIdFromOptions = context.getPipelineOptions().getProject(); - LOG.warn(String.format(BigQueryIO.SET_PROJECT_FROM_OPTIONS_WARNING, table.getDatasetId(), - table.getDatasetId(), table.getTableId(), projectIdFromOptions)); - table.setProjectId(projectIdFromOptions); - } - - context.addInput(PropertyNames.BIGQUERY_TABLE, table.getTableId()); - context.addInput(PropertyNames.BIGQUERY_DATASET, table.getDatasetId()); - if (table.getProjectId() != null) { - context.addInput(PropertyNames.BIGQUERY_PROJECT, table.getProjectId()); - } - } - context.addValueOnlyOutput(PropertyNames.OUTPUT, context.getOutput(transform)); - } - } -}