()
| 77 | |
| 78 | |
| 79 | def load(): |
| 80 | # As of this writing, Impala isn't able to write nested data in parquet format. |
| 81 | # Instead, the data will be written in text format, then Hive will be used to |
| 82 | # convert from text to parquet. |
| 83 | |
| 84 | with cluster.impala.cursor() as impala: |
| 85 | impala.ensure_empty_db(target_db) |
| 86 | impala.execute("USE %s" % target_db) |
| 87 | external = "" |
| 88 | tblproperties = "'{0}' = '{1}'".format(compression_key, compression_value) |
| 89 | # For Hive 3+, workaround for HIVE-22371 (CTAS puts files in the wrong place) by |
| 90 | # explicitly creating an external table so that files are in the external warehouse |
| 91 | # directory. Use external.table.purge=true so that it is equivalent to a Hive 2 |
| 92 | # managed table. |
| 93 | # For Apache Hive, HIVE-20085 (Hive 4) Allow CTAS. |
| 94 | if HIVE_MAJOR_VERSION >= 3 and os.environ["USE_APACHE_HIVE_3"] != "true": |
| 95 | external = "EXTERNAL" |
| 96 | tblproperties += ",'external.table.purge'='TRUE'" |
| 97 | sql_params = { |
| 98 | "source_db": source_db, |
| 99 | "target_db": target_db, |
| 100 | "file_format": file_format, |
| 101 | "chunks": chunks, |
| 102 | "warehouse_dir": cluster.hive.warehouse_dir, |
| 103 | "tblproperties": tblproperties, |
| 104 | "external": external} |
| 105 | |
| 106 | # Split table creation into multiple queries or "chunks" so less memory is needed. |
| 107 | for chunk_idx in range(chunks): |
| 108 | sql_params["chunk_idx"] = chunk_idx |
| 109 | |
| 110 | # Create the nested data in text format. The \00#'s are nested field terminators, |
| 111 | # where the numbers correspond to the nesting level. |
| 112 | tmp_orders_sql = r""" |
| 113 | SELECT STRAIGHT_JOIN |
| 114 | o_orderkey, o_custkey, o_orderstatus, o_totalprice, o_orderdate, |
| 115 | o_orderpriority, o_clerk, o_shippriority, o_comment, |
| 116 | GROUP_CONCAT( |
| 117 | CONCAT( |
| 118 | CAST(l_partkey AS STRING), '\005', |
| 119 | CAST(l_suppkey AS STRING), '\005', |
| 120 | CAST(l_linenumber AS STRING), '\005', |
| 121 | CAST(l_quantity AS STRING), '\005', |
| 122 | CAST(l_extendedprice AS STRING), '\005', |
| 123 | CAST(l_discount AS STRING), '\005', |
| 124 | CAST(l_tax AS STRING), '\005', |
| 125 | CAST(l_returnflag AS STRING), '\005', |
| 126 | CAST(l_linestatus AS STRING), '\005', |
| 127 | CAST(l_shipdate AS STRING), '\005', |
| 128 | CAST(l_commitdate AS STRING), '\005', |
| 129 | CAST(l_receiptdate AS STRING), '\005', |
| 130 | CAST(l_shipinstruct AS STRING), '\005', |
| 131 | CAST(l_shipmode AS STRING), '\005', |
| 132 | CAST(l_comment AS STRING) |
| 133 | ), '\004' |
| 134 | ) AS lineitems_string |
| 135 | FROM {source_db}.lineitem |
| 136 | INNER JOIN [SHUFFLE] {source_db}.orders ON o_orderkey = l_orderkey |
no test coverage detected