feat(tunnel-examples): 改进 Schema Evolution 示例

Author: dingxin-tech

odps-examples/tunnel-examples/src/main/java/SchemaEvolution/StreamUploadIfSchemaEvolutionExpectedSample.java

@@ -1,7 +1,10 @@
 package SchemaEvolution;
 
+import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
+import com.aliyun.odps.Column;
 import com.aliyun.odps.Instance;
 import com.aliyun.odps.Odps;
 import com.aliyun.odps.OdpsException;
@@ -108,7 +111,8 @@ public static void main(String[] args) throws Exception {
           // IllegalArgumentException will throw when the data is not compatible with the session schema
           // then should rebuild the session
           recordPack.flush();
-          rebuildSessionUtilSchemaEvolution(odps, null);
+          session = rebuildSessionUtilSchemaEvolution(odps, null);
+          recordPack = session.newRecordPack();
         }
         // Append the record to the record pack for uploading
         recordPack.append(record);
@@ -146,8 +150,11 @@ private static TableTunnel.StreamUploadSession rebuildSessionUtilSchemaEvolution
                 .setPartitionSpec(partition == null ? null : new PartitionSpec(partition))
                 .allowSchemaMismatch(false)
                 .build();
-      } while (!odps.tables().get(project, table).getSchema()
-          .basicallyEquals(session.getSchema()));
+        System.out.println("Session Schema: " + debugString(session.getSchema()));
+        System.out.println("Table Schema: " + debugString(odps.tables().get(project, table).getSchema()));
+
+      } while (!basicallyEquals(odps.tables().get(project, table).getSchema()
+          , session.getSchema()));
       return session;
     }
   }
@@ -179,4 +186,30 @@ private static void triggerSchemaEvolution() throws OdpsException {
     System.out.println(getOdps().logview().generateLogView(instance, 24));
     instance.waitForSuccess();
   }
+
+  private static String debugString(TableSchema schema) {
+    return schema.getAllColumns().stream()
+        .map(column -> column.getName() + "(" + column.getTypeInfo().getTypeName() + ")")
+        .collect(Collectors.joining(", "));
+  }
+
+  /**
+   * Check if two schemas are basically equal
+   */
+  private static boolean basicallyEquals(TableSchema a, TableSchema b) {
+    List<Column> columnsA = a.getAllColumns();
+    List<Column> columnsB = b.getAllColumns();
+    if (columnsA.size() != columnsB.size()) {
+      return false;
+    }
+    for (int i = 0; i < columnsA.size(); i++) {
+      Column columnA = columnsA.get(i);
+      Column columnB = columnsB.get(i);
+      if (!columnA.getName().equals(columnB.getName()) || !columnA.getTypeInfo().getTypeName()
+          .equals(columnB.getTypeInfo().getTypeName())) {
+        return false;
+      }
+    }
+    return true;
+  }
 }

odps-examples/tunnel-examples/src/main/java/SchemaEvolution/StreamUploadIfSchemaEvolutionUnexpectedSample.java

@@ -1,7 +1,10 @@
 package SchemaEvolution;
 
+import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
+import com.aliyun.odps.Column;
 import com.aliyun.odps.Instance;
 import com.aliyun.odps.Odps;
 import com.aliyun.odps.OdpsException;
@@ -158,8 +161,11 @@ private static TableTunnel.StreamUploadSession rebuildSessionUtilSchemaEvolution
                 .setPartitionSpec(partition == null ? null : new PartitionSpec(partition))
                 .allowSchemaMismatch(false)
                 .build();
-      } while (!odps.tables().get(project, table).getSchema()
-          .basicallyEquals(session.getSchema()));
+        System.out.println("Session Schema: " + debugString(session.getSchema()));
+        System.out.println("Table Schema: " + debugString(odps.tables().get(project, table).getSchema()));
+
+      } while (!basicallyEquals(odps.tables().get(project, table).getSchema()
+          , session.getSchema()));
       return session;
     }
   }
@@ -191,4 +197,30 @@ private static void triggerSchemaEvolution() throws OdpsException {
     System.out.println(getOdps().logview().generateLogView(instance, 24));
     instance.waitForSuccess();
   }
+
+  private static String debugString(TableSchema schema) {
+    return schema.getAllColumns().stream()
+        .map(column -> column.getName() + "(" + column.getTypeInfo().getTypeName() + ")")
+        .collect(Collectors.joining(", "));
+  }
+
+  /**
+   * Check if two schemas are basically equal
+   */
+  private static boolean basicallyEquals(TableSchema a, TableSchema b) {
+    List<Column> columnsA = a.getAllColumns();
+    List<Column> columnsB = b.getAllColumns();
+    if (columnsA.size() != columnsB.size()) {
+      return false;
+    }
+    for (int i = 0; i < columnsA.size(); i++) {
+      Column columnA = columnsA.get(i);
+      Column columnB = columnsB.get(i);
+      if (!columnA.getName().equals(columnB.getName()) || !columnA.getTypeInfo().getTypeName()
+          .equals(columnB.getTypeInfo().getTypeName())) {
+        return false;
+      }
+    }
+    return true;
+  }
 }

odps-examples/tunnel-examples/src/main/java/TunnelStreamingUploadExample.java

@@ -1,6 +1,11 @@
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 
+import com.aliyun.odps.Instance;
+import com.aliyun.odps.Odps;
+import com.aliyun.odps.OdpsException;
+import com.aliyun.odps.PartitionSpec;
+import com.aliyun.odps.TableSchema;
 import com.aliyun.odps.account.Account;
 import com.aliyun.odps.account.AliyunAccount;
 import com.aliyun.odps.account.StsAccount;

odps-examples/tunnel-examples/src/main/java/TunnelTimeDataSample.java

@@ -16,6 +16,9 @@
 import com.aliyun.odps.TableSchema;
 import com.aliyun.odps.account.Account;
 import com.aliyun.odps.account.AliyunAccount;
+import com.aliyun.odps.data.ArrayRecord;
+import com.aliyun.odps.data.RecordReader;
+import com.aliyun.odps.data.RecordWriter;
 import com.aliyun.odps.tunnel.TableTunnel;
 import com.aliyun.odps.tunnel.TableTunnel.DownloadSession;
 import com.aliyun.odps.tunnel.TableTunnel.UploadSession;