Runs COMPUTE STATS over the selected tables. The target tables can be filtered by specifying a list of databases and/or table names. If no filters are specified this will run COMPUTE STATS on all tables in all databases. parallelism controls the size of the thread pool to which compute_sta
(client_factory, db_names=None, table_names=None,
exclude_table_names=None, continue_on_error=False,
parallelism=DEFAULT_PARALLELISM)
| 62 | |
| 63 | |
| 64 | def compute_stats(client_factory, db_names=None, table_names=None, |
| 65 | exclude_table_names=None, continue_on_error=False, |
| 66 | parallelism=DEFAULT_PARALLELISM): |
| 67 | """ |
| 68 | Runs COMPUTE STATS over the selected tables. The target tables can be filtered by |
| 69 | specifying a list of databases and/or table names. If no filters are specified this will |
| 70 | run COMPUTE STATS on all tables in all databases. |
| 71 | |
| 72 | parallelism controls the size of the thread pool to which compute_stats |
| 73 | is sent. |
| 74 | """ |
| 75 | LOG.info("Enumerating databases and tables for compute stats. " |
| 76 | "db_names={} table_names={} exclude_table_names={} parallelism={}.".format( |
| 77 | str(db_names), str(table_names), str(exclude_table_names), parallelism |
| 78 | )) |
| 79 | |
| 80 | pool = multiprocessing.pool.ThreadPool(processes=parallelism) |
| 81 | futures = [] |
| 82 | |
| 83 | with client_factory() as impala_client: |
| 84 | db_table_map = {} |
| 85 | total_tables = 0 |
| 86 | all_dbs = set(name.split('\t')[0].lower() for name |
| 87 | in impala_client.execute("show databases").data) |
| 88 | selected_dbs = all_dbs if db_names is None else set(db_names) |
| 89 | for db in all_dbs.intersection(selected_dbs): |
| 90 | all_tables = set( |
| 91 | [t.lower() for t in impala_client.execute("show tables in %s" % db).data]) |
| 92 | selected_tables = all_tables if table_names is None else set(table_names) |
| 93 | excluded_tables = (set() if exclude_table_names is None |
| 94 | else set(exclude_table_names)) |
| 95 | tables_to_compute = (all_tables.intersection(selected_tables) |
| 96 | - excluded_tables) |
| 97 | db_table_map[db] = tables_to_compute |
| 98 | total_tables += len(tables_to_compute) |
| 99 | |
| 100 | for db, tables in db_table_map.items(): |
| 101 | for table in tables: |
| 102 | # Submit command to threadpool |
| 103 | futures.append( |
| 104 | pool.apply_async(compute_stats_table, (client_factory, db, table,))) |
| 105 | |
| 106 | # Wait for all stats commands to finish |
| 107 | completed = 0 |
| 108 | for f in futures: |
| 109 | try: |
| 110 | f.get() |
| 111 | completed += 1 |
| 112 | except Exception as e: |
| 113 | if not continue_on_error: |
| 114 | log_completion(completed, total_tables, e) |
| 115 | raise e |
| 116 | log_completion(completed, total_tables) |
| 117 | pool.terminate() |
| 118 | |
| 119 | |
| 120 | if __name__ == "__main__": |
no test coverage detected