public void runTemplate()

in java/src/main/java/com/google/cloud/dataproc/templates/gcs/GCStoBigquery.java [81:125]

38 lines of code
6 McCabe index (conditional complexity)


  public void runTemplate() {

    SparkSession spark = SparkSession.builder().appName("GCS to Bigquery load").getOrCreate();
    LOGGER.info("input format: {}", inputFileFormat);

    // Set log level
    spark.sparkContext().setLogLevel(sparkLogLevel);

    Dataset<Row> inputData = null;

    switch (inputFileFormat) {
      case GCS_BQ_CSV_FORMAT:
        inputData =
            spark
                .read()
                .format(GCS_BQ_CSV_FORMAT)
                .option(GCS_BQ_CSV_HEADER, true)
                .option(GCS_BQ_CSV_INFOR_SCHEMA, true)
                .load(inputFileLocation);
        break;
      case GCS_BQ_AVRO_FORMAT:
        inputData = spark.read().format(GCS_BQ_AVRO_EXTD_FORMAT).load(inputFileLocation);
        break;
      case GCS_BQ_PRQT_FORMAT:
        inputData = spark.read().parquet(inputFileLocation);
        break;
      default:
        throw new IllegalArgumentException(
            "Currently avro, parquet and csv are the only supported formats");
    }

    if (bqTempTable != null && bqTempQuery != null) {
      inputData.createOrReplaceGlobalTempView(bqTempTable);
      inputData = spark.sql(bqTempQuery);
    }

    inputData
        .write()
        .format(GCS_BQ_OUTPUT_FORMAT)
        .option(GCS_BQ_CSV_HEADER, true)
        .option(GCS_BQ_OUTPUT, bigQueryDataset + "." + bigQueryTable)
        .option(GCS_BQ_TEMP_BUCKET, bqTempBucket)
        .mode(SaveMode.valueOf(outputMode))
        .save();
  }