1. Fix bug partition not migrated. 2. Fix OutOfMemoryException due to coalesce

Author: dichenli

utilities/Hive_metastore_migration/src/export_from_datacatalog.py

@@ -27,9 +27,9 @@ def transform_catalog_to_df(dyf):
 def datacatalog_migrate_to_s3(databases, tables, partitions, output_path):
 
     # load
-    coalesce_by_row_count(databases).write.format('json').mode('overwrite').save(output_path + 'databases')
-    coalesce_by_row_count(tables).write.format('json').mode('overwrite').save(output_path + 'tables')
-    coalesce_by_row_count(partitions).write.format('json').mode('overwrite').save(output_path + 'partitions')
+    databases.write.format('json').mode('overwrite').save(output_path + 'databases')
+    tables.write.format('json').mode('overwrite').save(output_path + 'tables')
+    partitions.write.format('json').mode('overwrite').save(output_path + 'partitions')
 
 
 # apply hard-coded schema on dataframes, ensure schema is consistent for transformations

utilities/Hive_metastore_migration/src/hive_metastore_migration.py

@@ -225,20 +225,6 @@ def join_other_to_single_column(df, other, on, how, new_column_name):
     return df.join(other=other_combined, on=on, how=how)
 
 
-def coalesce_by_row_count(df, desired_rows_per_partition=10):
-    """
-    Coalesce dataframe to reduce number of partitions, to avoid fragmentation of data
-    :param df: dataframe
-    :param desired_rows_per_partition: desired number of rows per partition, there is no guarantee the actual rows count
-    is larger or smaller
-    :type df: DataFrame
-    :return: dataframe coalesced
-    """
-    count = df.count()
-    partitions = count / desired_rows_per_partition + 1
-    return df.coalesce(partitions)
-
-
 def batch_items_within_partition(sql_context, df, key_col, value_col, values_col):
     """
     Group a DataFrame of key, value pairs, create a list of values for the same key in each spark partition, but there
@@ -1432,9 +1418,9 @@ def etl_from_metastore(sc, sql_context, db_prefix, table_prefix, hive_metastore,
     # load
     output_path = get_output_dir(options['output_path'])
 
-    coalesce_by_row_count(databases).write.format('json').mode('overwrite').save(output_path + 'databases')
-    coalesce_by_row_count(tables).write.format('json').mode('overwrite').save(output_path + 'tables')
-    coalesce_by_row_count(partitions).write.format('json').mode('overwrite').save(output_path + 'partitions')
+    databases.write.format('json').mode('overwrite').save(output_path + 'databases')
+    tables.write.format('json').mode('overwrite').save(output_path + 'tables')
+    partitions.write.format('json').mode('overwrite').save(output_path + 'partitions')
 
 
 def etl_to_metastore(sc, sql_context, hive_metastore, options):

utilities/Hive_metastore_migration/src/import_into_datacatalog.py

@@ -35,11 +35,8 @@ def transform_df_to_catalog_import_schema(sql_context, glue_context, df_database
 
 def import_datacatalog(sql_context, glue_context, datacatalog_name, databases, tables, partitions, region):
 
-    # TEMP: get around datacatalog writer performance issue
-    limited_partitions = partitions.limit(10)
-
     (dyf_databases, dyf_tables, dyf_partitions) = transform_df_to_catalog_import_schema(
-        sql_context, glue_context, databases, tables, limited_partitions)
+        sql_context, glue_context, databases, tables, partitions)
 
     # load
     glue_context.write_dynamic_frame.from_options(