liufuyang
commented
5 years ago This is the project pom file: (basically changed based generated from https://cloud.google.com/dataflow/docs/quickstarts/quickstart-java-maven guide) https://pastebin.com/UmmihvYX
This is the command I used to run the job:
mvn clean compile
export GOOGLE_APPLICATION_CREDENTIALS=$(pwd)/key.json
mvn -Pdataflow-runner exec:java \
-Dexec.mainClass=org.apache.beam.examples.SourceRowCount \
-Dexec.args="--project=promotions-targeting \
--region=europe-west1 \
--stagingLocation=gs://promo-targeting-temporary/fuyangl/staging/ \
--tempLocation=gs://promo-targeting-temporary/fuyangl/temp/ \
--output=gs://promo-targeting-temporary/fuyangl/bt-count-output \
--bigtableProjectId=draper \
--bigtableInstanceId=draper \
--bigtableTableId=cohorts \
--numWorkers=3 \
--workerMachineType=n1-standard-8 \
--serviceAccount=fuyangl@promotions-targeting.iam.gserviceaccount.com \
--runner=DataflowRunner"And this is the job java code:
package org.apache.beam.examples;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.Read;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import com.google.cloud.bigtable.beam.CloudBigtableIO;
import com.google.cloud.bigtable.beam.CloudBigtableScanConfiguration;
public class SourceRowCount {
public interface CloudBigtableOptions extends PipelineOptions {
@Description("The Google Cloud project ID for the Cloud Bigtable instance.")
String getProjectId();
void setProjectId(String projectId);
@Description("The Google Cloud project ID for the Cloud Bigtable instance.")
String getBigtableProjectId();
void setBigtableProjectId(String bigtableProjectId);
@Description("The Google Cloud Bigtable instance ID .")
String getBigtableInstanceId();
void setBigtableInstanceId(String bigtableInstanceId);
@Description("The Cloud Bigtable table ID in the instance." )
String getBigtableTableId();
void setBigtableTableId(String bigtableTableId);
/** Set this required option to specify where to write the output. */
@Description("Path of the file to write to")
@Validation.Required
String getOutput();
void setOutput(String value);
}
// Converts a Long to a String so that it can be written to a file.
static DoFn<Long, String> stringifier = new DoFn<Long, String>() {
private static final long serialVersionUID = 1L;
@ProcessElement
public void processElement(DoFn<Long, String>.ProcessContext context) throws Exception {
context.output(context.element().toString());
}
};
public static void main(String[] args) {
CloudBigtableOptions options =
PipelineOptionsFactory.fromArgs(args).withValidation().as(CloudBigtableOptions.class);
String PROJECT_ID = options.getBigtableProjectId();
String INSTANCE_ID = options.getBigtableInstanceId();
String TABLE_ID = options.getBigtableTableId();
// [START bigtable_dataflow_connector_scan_config]
Scan scan = new Scan();
scan.setCacheBlocks(false);
scan.setFilter(new FirstKeyOnlyFilter());
// CloudBigtableTableConfiguration contains the project, zone, cluster and table to connect to.
// You can supply an optional Scan() to filter the rows that will be read.
CloudBigtableScanConfiguration config =
new CloudBigtableScanConfiguration.Builder()
.withProjectId(PROJECT_ID)
.withInstanceId(INSTANCE_ID)
.withTableId(TABLE_ID)
.withScan(scan)
.build();
Pipeline p = Pipeline.create(options);
p.apply(Read.from(CloudBigtableIO.read(config)))
.apply(Count.<Result>globally())
.apply(ParDo.of(stringifier))
.apply(TextIO.write().to(options.getOutput()));
// [END bigtable_dataflow_connector_scan_config]
p.run().waitUntilFinish();
// Once this is done, you can get the result file via "gsutil cp <resultLocation>-00000-of-00001"
}
}
rahulKQL
sduskis
johnlhamilton
igorbernstein2
Basically I am trying to follow the guide from
And I wanted to run the job
SourceRowCountto count the bigtable row numbers.As I used the pom.xml from the second url above, some of the packages are newer than the first url above.
Then in encountered this issue: https://github.com/googleapis/cloud-bigtable-client/issues/2150
Then I used the solution mentioned in the #2150 to change
bigtable-hbase-beamto 1.5.0, the local error ofcom.google.bigtable.repackaged.io.grpc.StatusRuntimeException: UNAUTHENTICATED: Request had invalid authentication credentials.is gone.However I still have
Permission deniederror on the log of dataflow job while scan bigtable. I believed that I have set the service accounts correct for my project.Then I changed
<beam.version>2.13.0</beam.version>to<beam.version>2.11.0</beam.version>, my job stopped reportingPermission deninedissue on dataflow. However the job still failed after a long hour with message:Workflow failed. Causes: The Dataflow job appears to be stuck because no worker activity has been seen in the last 1h.So my question now is: How can I do a bigtable row count properly. Should I just use the exact configuration of the
SourceRowCountexample or?