akj009

spark2-submit --class com.mptyminds.dataflow.Main \
	--master yarn --deploy-mode client \
	--driver-memory 2g --executor-memory 1g --executor-cores 1 \
	--conf spark.yarn.appMasterEnv.GOOGLE_APPLICATION_CREDENTIALS=<credenial_file_path> \
	--conf spark.yarn.executorEnv.GOOGLE_APPLICATION_CREDENTIALS=<credenial_file_path> \
	--conf spark.executorEnv.GOOGLE_APPLICATION_CREDENTIALS=<credenial_file_path> \
	/path/to/final-shaded.jar \
	--hdfsConfiguration=[{\"fs.default.name\":\"hdfs:/host:port\"}] \
	--sparkMaster=yarn --streaming=false \
	--project=<gcp-project> \

akj009

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.mptyminds.dataflow</groupId>
    <artifactId>bigquery-to-hdfs</artifactId>
    <version>1.0</version>

akj009

genericRecordPCollection.apply(
      FileIO
          .<GenericRecord>write()
          .via(ParquetIO.sink(schema))
          .to(pipelineOptions.getOutputFilePath())
          .withNumShards(pipelineOptions.getOutputFileShardCount())
          .withNaming((window, pane, numShards, shardIndex, compression) -> String.format(
                  "%s_%d_%d.%s", outputFilePrefix, startTime, endTime, "parq"))
);

akj009

final PCollection<GenericRecord> genericRecordPCollection = pipeline.apply(
                BigQueryIO
                        .read(SchemaAndRecord::getRecord)
                        .fromQuery(query)
                        .withCoder(AvroCoder.of(schema)).usingStandardSql()
);