Upgrade beam to 2.16.0 (#28)

Author: yu-iskw

README.md

@@ -46,7 +46,7 @@ Apache Beam's `DatastoreIO` doesn't allow us to write same key at once.
 mvn clean package
 
 # Run bigquery-to-datastore via the compiled JAR file
-java -cp $(pwd)/target/bigquery-to-datastore-bundled-0.5.2.jar \
+java -cp $(pwd)/target/bigquery-to-datastore-bundled-0.6.0.jar \
   com.github.yuiskw.beam.BigQuery2Datastore \
   --project=your-gcp-project \
   --runner=DataflowRunner \
@@ -71,7 +71,7 @@ mvn clean package
 make package
 
 # run
-java -cp $(pwd)/target/bigquery-to-datastore-bundled-0.5.2.jar --help
+java -cp $(pwd)/target/bigquery-to-datastore-bundled-0.6.0.jar --help
 # or
 ./bin/bigquery-to-datastore --help
 ```
@@ -80,7 +80,7 @@ java -cp $(pwd)/target/bigquery-to-datastore-bundled-0.5.2.jar --help
 We also offers docker images for this project in [yuiskw/bigquery\-to\-datastore \- Docker Hub](https://hub.docker.com/r/yuiskw/bigquery-to-datastore/).
 We have several docker images based on Apache Beam versions.
 ```
-docker run yuiskw/bigquery-to-datastore:0.5.2-beam-2.1 --help
+docker run yuiskw/bigquery-to-datastore:0.6.0-beam-2.16.0 --help
 ```
 
 ### How to install it with homebrew

pom.xml

@@ -22,28 +22,31 @@
 
   <groupId>com.github.yuiskw</groupId>
   <artifactId>bigquery-to-datastore</artifactId>
-  <version>0.5.2</version>
+  <version>0.6.0</version>
 
   <packaging>jar</packaging>
 
   <properties>
-    <beam.version>2.1.0</beam.version>
+    <beam.version>2.16.0</beam.version>
 
-    <bigquery.version>v2-rev295-1.22.0</bigquery.version>
-    <google-clients.version>1.22.0</google-clients.version>
-    <guava.version>20.0</guava.version>
-    <hamcrest.version>1.3</hamcrest.version>
-    <jackson.version>2.8.9</jackson.version>
-    <joda.version>2.4</joda.version>
-    <junit.version>4.12</junit.version>
-    <maven-compiler-plugin.version>3.6.1</maven-compiler-plugin.version>
-    <maven-exec-plugin.version>1.4.0</maven-exec-plugin.version>
+    <bigquery.version>v2-rev20181104-1.27.0</bigquery.version>
+    <google-clients.version>1.27.0</google-clients.version>
+    <hamcrest.version>2.1</hamcrest.version>
+    <jackson.version>2.9.10</jackson.version>
+    <joda.version>2.10.3</joda.version>
+    <junit.version>4.13-beta-3</junit.version>
+    <maven-compiler-plugin.version>3.7.0</maven-compiler-plugin.version>
+    <maven-exec-plugin.version>1.6.0</maven-exec-plugin.version>
     <maven-jar-plugin.version>3.0.2</maven-jar-plugin.version>
-    <maven-shade-plugin.version>3.0.0</maven-shade-plugin.version>
-    <pubsub.version>v1-rev10-1.22.0</pubsub.version>
-    <slf4j.version>1.7.14</slf4j.version>
-    <spark.version>1.6.3</spark.version>
-    <surefire-plugin.version>2.20</surefire-plugin.version>
+    <maven-shade-plugin.version>3.1.0</maven-shade-plugin.version>
+    <mockito.version>3.0.0</mockito.version>
+    <pubsub.version>v1-rev20181105-1.27.0</pubsub.version>
+    <slf4j.version>1.7.25</slf4j.version>
+    <spark.version>2.4.4</spark.version>
+    <hadoop.version>2.7.3</hadoop.version>
+    <maven-surefire-plugin.version>2.21.0</maven-surefire-plugin.version>
+    <nemo.version>0.1</nemo.version>
+    <flink.artifact.name>beam-runners-flink-1.8</flink.artifact.name>
   </properties>
 
   <repositories>
@@ -75,7 +78,7 @@
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-surefire-plugin</artifactId>
-        <version>${surefire-plugin.version}</version>
+        <version>${maven-surefire-plugin.version}</version>
         <configuration>
           <parallel>all</parallel>
           <threadCount>4</threadCount>
@@ -85,7 +88,7 @@
           <dependency>
             <groupId>org.apache.maven.surefire</groupId>
             <artifactId>surefire-junit47</artifactId>
-            <version>${surefire-plugin.version}</version>
+            <version>${maven-surefire-plugin.version}</version>
           </dependency>
         </dependencies>
       </plugin>
@@ -166,54 +169,34 @@
     </profile>
 
     <profile>
-      <id>dataflow-runner</id>
-      <!-- Makes the DataflowRunner available when running a pipeline. -->
+      <id>portable-runner</id>
+      <activation>
+        <activeByDefault>true</activeByDefault>
+      </activation>
+      <!-- Makes the PortableRunner available when running a pipeline. -->
       <dependencies>
         <dependency>
           <groupId>org.apache.beam</groupId>
-          <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
+          <artifactId>beam-runners-reference-java</artifactId>
           <version>${beam.version}</version>
+          <scope>runtime</scope>
         </dependency>
       </dependencies>
     </profile>
 
     <profile>
-      <id>spark-runner</id>
-      <!-- Makes the SparkRunner available when running a pipeline. Additionally,
-           overrides some Spark dependencies to Beam-compatible versions. -->
+      <id>dataflow-runner</id>
+      <!-- Makes the DataflowRunner available when running a pipeline. -->
       <dependencies>
         <dependency>
           <groupId>org.apache.beam</groupId>
-          <artifactId>beam-runners-spark</artifactId>
-          <version>${beam.version}</version>
-          <scope>runtime</scope>
-        </dependency>
-        <dependency>
-          <groupId>org.apache.beam</groupId>
-          <artifactId>beam-sdks-java-io-hadoop-file-system</artifactId>
+          <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
           <version>${beam.version}</version>
           <scope>runtime</scope>
         </dependency>
-        <dependency>
-          <groupId>org.apache.spark</groupId>
-          <artifactId>spark-streaming_2.10</artifactId>
-          <version>${spark.version}</version>
-          <scope>runtime</scope>
-          <exclusions>
-            <exclusion>
-              <groupId>org.slf4j</groupId>
-              <artifactId>jul-to-slf4j</artifactId>
-            </exclusion>
-          </exclusions>
-        </dependency>
-        <dependency>
-          <groupId>com.fasterxml.jackson.module</groupId>
-          <artifactId>jackson-module-scala_2.10</artifactId>
-          <version>${jackson.version}</version>
-          <scope>runtime</scope>
-        </dependency>
       </dependencies>
     </profile>
+
   </profiles>
 
   <dependencies>
@@ -300,12 +283,6 @@
       <version>${joda.version}</version>
     </dependency>
 
-    <dependency>
-      <groupId>com.google.guava</groupId>
-      <artifactId>guava</artifactId>
-      <version>${guava.version}</version>
-    </dependency>
-
     <!-- Add slf4j API frontend binding with JUL backend -->
     <dependency>
       <groupId>org.slf4j</groupId>
@@ -325,7 +302,13 @@
          which is used in the main code of DebuggingWordCount example. -->
     <dependency>
       <groupId>org.hamcrest</groupId>
-      <artifactId>hamcrest-all</artifactId>
+      <artifactId>hamcrest-core</artifactId>
+      <version>${hamcrest.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-library</artifactId>
       <version>${hamcrest.version}</version>
     </dependency>
 
@@ -343,5 +326,11 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <version>${mockito.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>

src/main/java/com/github/yuiskw/beam/BigQuery2Datastore.java

@@ -72,7 +72,10 @@ public static void main(String[] args) {
     List<String> indexedColumns = parseIndexedColumns(options.getIndexedColumns());
 
     // Input
-    TableReference tableRef = new TableReference().setDatasetId(datasetId).setTableId(tableId);
+    TableReference tableRef = new TableReference()
+        .setProjectId(projectId)
+        .setDatasetId(datasetId)
+        .setTableId(tableId);
     BigQueryIO.Read reader = BigQueryIO.read().from(tableRef);
 
     // Output

src/main/java/com/github/yuiskw/beam/TableRow2EntityFn.java

@@ -13,7 +13,6 @@
 
 import com.google.api.services.bigquery.model.TableRow;
 import com.google.datastore.v1.*;
-import com.google.protobuf.Timestamp;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.joda.time.LocalDate;
 import org.joda.time.LocalDateTime;
@@ -54,19 +53,20 @@ public TableRow2EntityFn(
 
   /**
    * Convert Date to Timestamp
+   * @return
    */
-  public static Timestamp toTimestamp(Date date) {
+  public static com.google.protobuf.Timestamp toTimestamp(Date date) {
     long millis = date.getTime();
-    Timestamp timestamp = Timestamp.newBuilder()
+    com.google.protobuf.Timestamp timestamp = com.google.protobuf.Timestamp.newBuilder()
         .setSeconds(millis / 1000)
         .setNanos((int) ((millis % 1000) * 1000000))
         .build();
     return timestamp;
   }
 
-  public static Timestamp toTimestamp(Instant instant) {
+  public static com.google.protobuf.Timestamp toTimestamp(Instant instant) {
     long second = instant.getEpochSecond();
-    Timestamp timestamp = Timestamp.newBuilder()
+    com.google.protobuf.Timestamp timestamp = com.google.protobuf.Timestamp.newBuilder()
         .setSeconds(second)
         .build();
     return timestamp;
@@ -130,24 +130,24 @@ else if (value instanceof java.lang.Double) {
     // }
     // TIMESTAMP
     else if (value instanceof org.joda.time.LocalDateTime) {
-      Timestamp timestamp = toTimestamp(((LocalDateTime) value).toLocalDate().toDate());
+      com.google.protobuf.Timestamp timestamp = toTimestamp(((LocalDateTime) value).toLocalDate().toDate());
       v = Value.newBuilder().setTimestampValue(timestamp)
           .setExcludeFromIndexes(isExcluded).build();
     }
     else if (value instanceof String && parseTimestamp((String) value) != null) {
       Instant instant = parseTimestamp((String) value);
-      Timestamp timestamp = toTimestamp(instant);
+      com.google.protobuf.Timestamp timestamp = toTimestamp(instant);
       v = Value.newBuilder().setTimestampValue(timestamp)
           .setExcludeFromIndexes(isExcluded).build();
     }
     // DATE
     else if (value instanceof org.joda.time.LocalDate) {
-      Timestamp timestamp = toTimestamp(((LocalDate) value).toDate());
+      com.google.protobuf.Timestamp timestamp = toTimestamp(((LocalDate) value).toDate());
       v = Value.newBuilder().setTimestampValue(timestamp)
           .setExcludeFromIndexes(isExcluded).build();
     } else if (value instanceof String && parseDate((String) value) != null) {
       Instant instant = parseDate((String) value);
-      Timestamp timestamp = toTimestamp(instant);
+      com.google.protobuf.Timestamp timestamp = toTimestamp(instant);
       v = Value.newBuilder().setTimestampValue(timestamp)
           .setExcludeFromIndexes(isExcluded).build();
     }