Executes the query files provided using the execution engine specified in parallel using the given thread pool. Aborts immediately if any execution encounters an error.
(thread_pool, query_files, execution_type, step_name)
| 274 | |
| 275 | |
| 276 | def exec_query_files_parallel(thread_pool, query_files, execution_type, step_name): |
| 277 | """Executes the query files provided using the execution engine specified |
| 278 | in parallel using the given thread pool. Aborts immediately if any execution |
| 279 | encounters an error.""" |
| 280 | assert(execution_type == 'impala' or execution_type == 'hive') |
| 281 | if len(query_files) == 0: return |
| 282 | if execution_type == 'impala': |
| 283 | execution_function = exec_impala_query_from_file |
| 284 | elif execution_type == 'hive': |
| 285 | execution_function = exec_hive_query_from_file_beeline |
| 286 | |
| 287 | LOG.info('Begin step "%s".' % step_name) |
| 288 | start_time = time.time() |
| 289 | for result in thread_pool.imap_unordered(execution_function, query_files): |
| 290 | if not result: |
| 291 | thread_pool.terminate() |
| 292 | sys.exit(1) |
| 293 | total_time = time.time() - start_time |
| 294 | LOG.info('End step "%s". Total time: %.2fs\n' % (step_name, total_time)) |
| 295 | |
| 296 | |
| 297 | def impala_exec_query_files_parallel(thread_pool, query_files, step_name): |
no outgoing calls
no test coverage detected