DataLoaderPipeline.kt

val resultTableSpec = TableReference()
        .apply {
            projectId = "GCP_PROJECT_NAME"
            datasetId = "BIGQUERY_DATASET_NAME"
            tableId = "BIGQUERY_TABLE_NAME"
        }

pipeline
  .apply("Read files", TextIO.read().from("gs://BUCKET_NAME/*.csv"))
  .apply("Filter headers", PipelineSteps.filterHeaderRow())
  .apply("Split lines", PipelineSteps.splitCsvLine())
  .apply("Map to entity", PipelineSteps.mapElementsToEntity())
  .apply("Map to table row", PipelineSteps.mapEntityToTableRow())
  .apply("Write to BigQuery",   PipelineSteps.writeToBigQuery(resultTableSpec))

pipeline.run()