sandboxws

Pipeline pipeline = Pipeline.create(options);

sandboxws

PGSimpleDataSource pgDataSource = getPostgresDataSource(options);

sandboxws

PCollection<HashMap<String, Object>> rows = pipeline.apply(
  "Read Albums from PG",
  JdbcIO.<HashMap<String, Object>>read()
    .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(pgDataSource))
    .withCoder(TableRowCoder.of())
    // Map ResultSet row to a HashMap
    .withRowMapper(new RowMapper<HashMap<String, Object>>() {
      @Override
      public HashMap<String, Object> mapRow(ResultSet resultSet) throws Exception {
        return TableRowMapper.asMap(resultSet, tableName, pkName);

sandboxws

dependencies {
    // Apache Beam
    compile group: 'org.apache.beam', name: 'beam-sdks-java-core', version:'2.9.0' // Core Java SDK
    compile group: 'org.apache.beam', name: 'beam-sdks-java-io-google-cloud-platform', version:'2.9.0' // Google Cloud IO
    compile group: 'org.apache.beam', name: 'beam-sdks-java-io-jdbc', version:'2.9.0' // JDBC IO
    runtime group: 'org.apache.beam', name: 'beam-runners-direct-java', version:'2.9.0' // Beam Direct Runner
    // Dataflow
    compile group: 'com.google.cloud.dataflow', name: 'google-cloud-dataflow-java-sdk-all', version: '2.5.0' // Dataflow Java SDK
    // Logging
    runtime group: 'org.slf4j', name: 'slf4j-jdk14', version:'1.7.25'

sandboxws

package com.sandboxws.beam.coders;

import com.google.gson.Gson;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.HashMap;
import org.apache.beam.sdk.coders.AtomicCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.values.TypeDescriptor;

sandboxws

package com.sandboxws.chinook.bigquery.schema;

import com.google.api.services.bigquery.model.TableSchema;
import com.sandboxws.bigquery.TableSchemaBuilder;

/**
 * Album BigQuery table schema.
 *
 * @author Ahmed El.Hussaini
 */

sandboxws

package com.sandboxws.chinook;

import com.google.api.services.bigquery.model.TableRow;
import com.sandboxws.beam.AppOptions;
import com.sandboxws.beam.coders.TableRowCoder;
import com.sandboxws.chinook.bigquery.schema.AlbumTableSchema;

import java.sql.ResultSet;
import java.util.HashMap;
import java.util.Map;

sandboxws

package com.sandboxws.chinook;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.TimeZone;

sandboxws

package com.sandboxws.bigquery;

import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableSchema;
import java.util.ArrayList;
import java.util.List;

/**
 * BigQuery table schema builder.
 *

sandboxws

package com.sandboxws.beam;

import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;

/**
 * App specific Apache Beam pipeline options.
 *
 * @author Ahmed Elhossaini