@Test public void testNoPropsSet() { HoodieWriteConfig config = getConfigBuilder(schemaStr).build(); List<Row> rows = DataSourceTestUtils.generateRandomRows(10); Dataset<Row> dataset = sqlContext.createDataFrame(rows, structType); try { HoodieDatasetBulkInsertHelper.prepareHoodieDatasetForBulkInsert(sqlContext, config, dataset, "testStructName", "testNamespace"); fail("Should have thrown exception"); } catch (Exception e) { } config = getConfigBuilder(schemaStr).withProps(getProps(false, false, true, true)).build(); rows = DataSourceTestUtils.generateRandomRows(10); dataset = sqlContext.createDataFrame(rows, structType); try { HoodieDatasetBulkInsertHelper.prepareHoodieDatasetForBulkInsert(sqlContext, config, dataset, "testStructName", "testNamespace"); fail("Should have thrown exception"); } catch (Exception e) { } config = getConfigBuilder(schemaStr).withProps(getProps(false, true, false, true)).build(); rows = DataSourceTestUtils.generateRandomRows(10); dataset = sqlContext.createDataFrame(rows, structType); try { HoodieDatasetBulkInsertHelper.prepareHoodieDatasetForBulkInsert(sqlContext, config, dataset, "testStructName", "testNamespace"); fail("Should have thrown exception"); } catch (Exception e) { } config = getConfigBuilder(schemaStr).withProps(getProps(false, true, true, false)).build(); rows = DataSourceTestUtils.generateRandomRows(10); dataset = sqlContext.createDataFrame(rows, structType); try { HoodieDatasetBulkInsertHelper.prepareHoodieDatasetForBulkInsert(sqlContext, config, dataset, "testStructName", "testNamespace"); fail("Should have thrown exception"); } catch (Exception e) { } }
public static Dataset<Row> prepareHoodieDatasetForBulkInsert(SQLContext sqlContext, HoodieWriteConfig config, Dataset<Row> rows, String structName, String recordNamespace) { List<Column> originalFields = Arrays.stream(rows.schema().fields()).map(f -> new Column(f.name())).collect(Collectors.toList()); TypedProperties properties = new TypedProperties(); properties.putAll(config.getProps()); String keyGeneratorClass = properties.getString(DataSourceWriteOptions.KEYGENERATOR_CLASS_OPT_KEY()); KeyGenerator keyGenerator = (KeyGenerator) ReflectionUtils.loadClass(keyGeneratorClass, properties); StructType structTypeForUDF = rows.schema(); sqlContext.udf().register(RECORD_KEY_UDF_FN, (UDF1<Row, String>) keyGenerator::getRecordKey, DataTypes.StringType); sqlContext.udf().register(PARTITION_PATH_UDF_FN, (UDF1<Row, String>) keyGenerator::getPartitionPath, DataTypes.StringType); final Dataset<Row> rowDatasetWithRecordKeys = rows.withColumn(HoodieRecord.RECORD_KEY_METADATA_FIELD, callUDF(RECORD_KEY_UDF_FN, org.apache.spark.sql.functions.struct( JavaConverters.collectionAsScalaIterableConverter(originalFields).asScala().toSeq()))); final Dataset<Row> rowDatasetWithRecordKeysAndPartitionPath = rowDatasetWithRecordKeys.withColumn(HoodieRecord.PARTITION_PATH_METADATA_FIELD, callUDF(PARTITION_PATH_UDF_FN, org.apache.spark.sql.functions.struct( JavaConverters.collectionAsScalaIterableConverter(originalFields).asScala().toSeq()))); Dataset<Row> rowDatasetWithHoodieColumns = rowDatasetWithRecordKeysAndPartitionPath.withColumn(HoodieRecord.COMMIT_TIME_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)) .withColumn(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)) .withColumn(HoodieRecord.FILENAME_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)); List<Column> orderedFields = Stream.concat(HoodieRecord.HOODIE_META_COLUMNS.stream().map(Column::new), originalFields.stream()).collect(Collectors.toList()); Dataset<Row> colOrderedDataset = rowDatasetWithHoodieColumns.select( JavaConverters.collectionAsScalaIterableConverter(orderedFields).asScala().toSeq()); return colOrderedDataset .sort(functions.col(HoodieRecord.PARTITION_PATH_METADATA_FIELD), functions.col(HoodieRecord.RECORD_KEY_METADATA_FIELD)) .coalesce(config.getBulkInsertShuffleParallelism()); }
HoodieDatasetBulkInsertHelper { public static Dataset<Row> prepareHoodieDatasetForBulkInsert(SQLContext sqlContext, HoodieWriteConfig config, Dataset<Row> rows, String structName, String recordNamespace) { List<Column> originalFields = Arrays.stream(rows.schema().fields()).map(f -> new Column(f.name())).collect(Collectors.toList()); TypedProperties properties = new TypedProperties(); properties.putAll(config.getProps()); String keyGeneratorClass = properties.getString(DataSourceWriteOptions.KEYGENERATOR_CLASS_OPT_KEY()); KeyGenerator keyGenerator = (KeyGenerator) ReflectionUtils.loadClass(keyGeneratorClass, properties); StructType structTypeForUDF = rows.schema(); sqlContext.udf().register(RECORD_KEY_UDF_FN, (UDF1<Row, String>) keyGenerator::getRecordKey, DataTypes.StringType); sqlContext.udf().register(PARTITION_PATH_UDF_FN, (UDF1<Row, String>) keyGenerator::getPartitionPath, DataTypes.StringType); final Dataset<Row> rowDatasetWithRecordKeys = rows.withColumn(HoodieRecord.RECORD_KEY_METADATA_FIELD, callUDF(RECORD_KEY_UDF_FN, org.apache.spark.sql.functions.struct( JavaConverters.collectionAsScalaIterableConverter(originalFields).asScala().toSeq()))); final Dataset<Row> rowDatasetWithRecordKeysAndPartitionPath = rowDatasetWithRecordKeys.withColumn(HoodieRecord.PARTITION_PATH_METADATA_FIELD, callUDF(PARTITION_PATH_UDF_FN, org.apache.spark.sql.functions.struct( JavaConverters.collectionAsScalaIterableConverter(originalFields).asScala().toSeq()))); Dataset<Row> rowDatasetWithHoodieColumns = rowDatasetWithRecordKeysAndPartitionPath.withColumn(HoodieRecord.COMMIT_TIME_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)) .withColumn(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)) .withColumn(HoodieRecord.FILENAME_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)); List<Column> orderedFields = Stream.concat(HoodieRecord.HOODIE_META_COLUMNS.stream().map(Column::new), originalFields.stream()).collect(Collectors.toList()); Dataset<Row> colOrderedDataset = rowDatasetWithHoodieColumns.select( JavaConverters.collectionAsScalaIterableConverter(orderedFields).asScala().toSeq()); return colOrderedDataset .sort(functions.col(HoodieRecord.PARTITION_PATH_METADATA_FIELD), functions.col(HoodieRecord.RECORD_KEY_METADATA_FIELD)) .coalesce(config.getBulkInsertShuffleParallelism()); } }
HoodieDatasetBulkInsertHelper { public static Dataset<Row> prepareHoodieDatasetForBulkInsert(SQLContext sqlContext, HoodieWriteConfig config, Dataset<Row> rows, String structName, String recordNamespace) { List<Column> originalFields = Arrays.stream(rows.schema().fields()).map(f -> new Column(f.name())).collect(Collectors.toList()); TypedProperties properties = new TypedProperties(); properties.putAll(config.getProps()); String keyGeneratorClass = properties.getString(DataSourceWriteOptions.KEYGENERATOR_CLASS_OPT_KEY()); KeyGenerator keyGenerator = (KeyGenerator) ReflectionUtils.loadClass(keyGeneratorClass, properties); StructType structTypeForUDF = rows.schema(); sqlContext.udf().register(RECORD_KEY_UDF_FN, (UDF1<Row, String>) keyGenerator::getRecordKey, DataTypes.StringType); sqlContext.udf().register(PARTITION_PATH_UDF_FN, (UDF1<Row, String>) keyGenerator::getPartitionPath, DataTypes.StringType); final Dataset<Row> rowDatasetWithRecordKeys = rows.withColumn(HoodieRecord.RECORD_KEY_METADATA_FIELD, callUDF(RECORD_KEY_UDF_FN, org.apache.spark.sql.functions.struct( JavaConverters.collectionAsScalaIterableConverter(originalFields).asScala().toSeq()))); final Dataset<Row> rowDatasetWithRecordKeysAndPartitionPath = rowDatasetWithRecordKeys.withColumn(HoodieRecord.PARTITION_PATH_METADATA_FIELD, callUDF(PARTITION_PATH_UDF_FN, org.apache.spark.sql.functions.struct( JavaConverters.collectionAsScalaIterableConverter(originalFields).asScala().toSeq()))); Dataset<Row> rowDatasetWithHoodieColumns = rowDatasetWithRecordKeysAndPartitionPath.withColumn(HoodieRecord.COMMIT_TIME_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)) .withColumn(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)) .withColumn(HoodieRecord.FILENAME_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)); List<Column> orderedFields = Stream.concat(HoodieRecord.HOODIE_META_COLUMNS.stream().map(Column::new), originalFields.stream()).collect(Collectors.toList()); Dataset<Row> colOrderedDataset = rowDatasetWithHoodieColumns.select( JavaConverters.collectionAsScalaIterableConverter(orderedFields).asScala().toSeq()); return colOrderedDataset .sort(functions.col(HoodieRecord.PARTITION_PATH_METADATA_FIELD), functions.col(HoodieRecord.RECORD_KEY_METADATA_FIELD)) .coalesce(config.getBulkInsertShuffleParallelism()); }  }
HoodieDatasetBulkInsertHelper { public static Dataset<Row> prepareHoodieDatasetForBulkInsert(SQLContext sqlContext, HoodieWriteConfig config, Dataset<Row> rows, String structName, String recordNamespace) { List<Column> originalFields = Arrays.stream(rows.schema().fields()).map(f -> new Column(f.name())).collect(Collectors.toList()); TypedProperties properties = new TypedProperties(); properties.putAll(config.getProps()); String keyGeneratorClass = properties.getString(DataSourceWriteOptions.KEYGENERATOR_CLASS_OPT_KEY()); KeyGenerator keyGenerator = (KeyGenerator) ReflectionUtils.loadClass(keyGeneratorClass, properties); StructType structTypeForUDF = rows.schema(); sqlContext.udf().register(RECORD_KEY_UDF_FN, (UDF1<Row, String>) keyGenerator::getRecordKey, DataTypes.StringType); sqlContext.udf().register(PARTITION_PATH_UDF_FN, (UDF1<Row, String>) keyGenerator::getPartitionPath, DataTypes.StringType); final Dataset<Row> rowDatasetWithRecordKeys = rows.withColumn(HoodieRecord.RECORD_KEY_METADATA_FIELD, callUDF(RECORD_KEY_UDF_FN, org.apache.spark.sql.functions.struct( JavaConverters.collectionAsScalaIterableConverter(originalFields).asScala().toSeq()))); final Dataset<Row> rowDatasetWithRecordKeysAndPartitionPath = rowDatasetWithRecordKeys.withColumn(HoodieRecord.PARTITION_PATH_METADATA_FIELD, callUDF(PARTITION_PATH_UDF_FN, org.apache.spark.sql.functions.struct( JavaConverters.collectionAsScalaIterableConverter(originalFields).asScala().toSeq()))); Dataset<Row> rowDatasetWithHoodieColumns = rowDatasetWithRecordKeysAndPartitionPath.withColumn(HoodieRecord.COMMIT_TIME_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)) .withColumn(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)) .withColumn(HoodieRecord.FILENAME_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)); List<Column> orderedFields = Stream.concat(HoodieRecord.HOODIE_META_COLUMNS.stream().map(Column::new), originalFields.stream()).collect(Collectors.toList()); Dataset<Row> colOrderedDataset = rowDatasetWithHoodieColumns.select( JavaConverters.collectionAsScalaIterableConverter(orderedFields).asScala().toSeq()); return colOrderedDataset .sort(functions.col(HoodieRecord.PARTITION_PATH_METADATA_FIELD), functions.col(HoodieRecord.RECORD_KEY_METADATA_FIELD)) .coalesce(config.getBulkInsertShuffleParallelism()); }  static Dataset<Row> prepareHoodieDatasetForBulkInsert(SQLContext sqlContext,
      HoodieWriteConfig config, Dataset<Row> rows, String structName, String recordNamespace); }
HoodieDatasetBulkInsertHelper { public static Dataset<Row> prepareHoodieDatasetForBulkInsert(SQLContext sqlContext, HoodieWriteConfig config, Dataset<Row> rows, String structName, String recordNamespace) { List<Column> originalFields = Arrays.stream(rows.schema().fields()).map(f -> new Column(f.name())).collect(Collectors.toList()); TypedProperties properties = new TypedProperties(); properties.putAll(config.getProps()); String keyGeneratorClass = properties.getString(DataSourceWriteOptions.KEYGENERATOR_CLASS_OPT_KEY()); KeyGenerator keyGenerator = (KeyGenerator) ReflectionUtils.loadClass(keyGeneratorClass, properties); StructType structTypeForUDF = rows.schema(); sqlContext.udf().register(RECORD_KEY_UDF_FN, (UDF1<Row, String>) keyGenerator::getRecordKey, DataTypes.StringType); sqlContext.udf().register(PARTITION_PATH_UDF_FN, (UDF1<Row, String>) keyGenerator::getPartitionPath, DataTypes.StringType); final Dataset<Row> rowDatasetWithRecordKeys = rows.withColumn(HoodieRecord.RECORD_KEY_METADATA_FIELD, callUDF(RECORD_KEY_UDF_FN, org.apache.spark.sql.functions.struct( JavaConverters.collectionAsScalaIterableConverter(originalFields).asScala().toSeq()))); final Dataset<Row> rowDatasetWithRecordKeysAndPartitionPath = rowDatasetWithRecordKeys.withColumn(HoodieRecord.PARTITION_PATH_METADATA_FIELD, callUDF(PARTITION_PATH_UDF_FN, org.apache.spark.sql.functions.struct( JavaConverters.collectionAsScalaIterableConverter(originalFields).asScala().toSeq()))); Dataset<Row> rowDatasetWithHoodieColumns = rowDatasetWithRecordKeysAndPartitionPath.withColumn(HoodieRecord.COMMIT_TIME_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)) .withColumn(HoodieRecord.COMMIT_SEQNO_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)) .withColumn(HoodieRecord.FILENAME_METADATA_FIELD, functions.lit("").cast(DataTypes.StringType)); List<Column> orderedFields = Stream.concat(HoodieRecord.HOODIE_META_COLUMNS.stream().map(Column::new), originalFields.stream()).collect(Collectors.toList()); Dataset<Row> colOrderedDataset = rowDatasetWithHoodieColumns.select( JavaConverters.collectionAsScalaIterableConverter(orderedFields).asScala().toSeq()); return colOrderedDataset .sort(functions.col(HoodieRecord.PARTITION_PATH_METADATA_FIELD), functions.col(HoodieRecord.RECORD_KEY_METADATA_FIELD)) .coalesce(config.getBulkInsertShuffleParallelism()); }  static Dataset<Row> prepareHoodieDatasetForBulkInsert(SQLContext sqlContext,
      HoodieWriteConfig config, Dataset<Row> rows, String structName, String recordNamespace);  }