| #!/usr/bin/env impala-python |
| |
| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| '''This module provides random data generation and database population. |
| |
| When this module is run directly for purposes of database population, the default is |
| to use a fixed seed for randomization. The result should be that the generated random |
| data is the same regardless of when or where the execution is done. |
| |
| ''' |
| |
| import os |
| from copy import deepcopy |
| from logging import getLogger |
| from random import choice, randint, seed |
| from time import time |
| |
| from tests.comparison.data_generator_mapred_common import ( |
| estimate_rows_per_reducer, |
| MB_PER_REDUCER, |
| serialize, |
| TextTableDataGenerator) |
| from tests.comparison.common import Column, Table |
| from tests.comparison.db_types import ( |
| Char, |
| Decimal, |
| EXACT_TYPES, |
| get_char_class, |
| get_decimal_class, |
| get_varchar_class, |
| String, |
| Timestamp, |
| TYPES, |
| VarChar) |
| from tests.comparison import db_connection |
| |
| LOG = getLogger(__name__) |
| |
| def index_tables_in_db_if_possible(cursor): |
| if not cursor.conn.supports_index_creation: |
| return |
| for table_name in cursor.list_table_names(): |
| LOG.info('Indexing %s on %s' % (table_name, cursor.db_type)) |
| cursor.index_table(table_name) |
| |
| |
| def migrate_db(src_cursor, dst_cursor, include_table_names=None): |
| '''Read table metadata and data from the source database and create a replica in |
| the destination database. For example, the Impala functional test database could |
| be copied into Postgresql. |
| ''' |
| for table_name in src_cursor.list_table_names(): |
| if include_table_names and table_name not in include_table_names: |
| continue |
| table = src_cursor.describe_table(table_name) |
| dst_cursor.create_table(table) |
| src_cursor.execute('SELECT * FROM ' + table_name) |
| while True: |
| rows = src_cursor.fetchmany(size=100) |
| if not rows: |
| break |
| sql = dst_cursor.make_insert_sql_from_data(table, rows) |
| dst_cursor.execute(sql) |
| index_tables_in_db_if_possible(dst_cursor) |
| |
| |
| class DbPopulator(object): |
| '''This class will populate a database with randomly generated data. The population |
| includes table creation and data generation. Table names are hard coded as |
| table_<table number>. |
| |
| ''' |
| |
| def __init__(self, db_engine=db_connection.IMPALA): |
| self.cluster = None |
| self.db_name = None |
| self.db_engine = db_engine |
| |
| self.min_col_count = None |
| self.max_col_count = None |
| self.min_row_count = None |
| self.max_row_count = None |
| self.allowed_storage_formats = None |
| self.randomization_seed = None |
| |
| def populate_db(self, table_count, postgresql_conn=None): |
| '''Create tables with a random number of cols. |
| |
| The given db_name must have already been created. |
| ''' |
| self.cluster.hdfs.ensure_home_dir() |
| hdfs = self.cluster.hdfs.create_client() |
| |
| table_and_generators = list() |
| for table_idx in xrange(table_count): |
| table = self._create_random_table( |
| 'table_%s' % (table_idx + 1), |
| self.min_col_count, |
| self.max_col_count, |
| self.allowed_storage_formats) |
| self._prepare_table_storage(table, self.db_name) |
| if table.storage_format == 'TEXTFILE': |
| text_table = table |
| else: |
| text_table = deepcopy(table) |
| text_table.name += '_text' |
| text_table.storage_format = 'TEXTFILE' |
| text_table.storage_location = None |
| text_table.schema_location = None |
| self._prepare_table_storage(text_table, self.db_name) |
| table_data_generator = TextTableDataGenerator() |
| table_data_generator.randomization_seed = self.randomization_seed |
| table_data_generator.table = text_table |
| table_data_generator.row_count = randint(self.min_row_count, self.max_row_count) |
| table_and_generators.append((table, table_data_generator)) |
| |
| self._run_data_generator_mr_job([g for _, g in table_and_generators], self.db_name) |
| |
| with self.cluster.hive.cursor(db_name=self.db_name) as cursor: |
| for table, table_data_generator in table_and_generators: |
| cursor.create_table(table) |
| text_table = table_data_generator.table |
| if postgresql_conn: |
| with postgresql_conn.cursor() as postgresql_cursor: |
| postgresql_cursor.create_table(table) |
| for data_file in hdfs.list(text_table.storage_location): |
| with hdfs.read(text_table.storage_location + '/' + data_file) as reader: |
| postgresql_cursor.copy_expert( |
| r"COPY %s FROM STDIN WITH DELIMITER E'\x01'" % table.name, reader) |
| if table.storage_format != 'TEXTFILE': |
| cursor.create_table(text_table) |
| cursor.execute('INSERT INTO %s SELECT * FROM %s' |
| % (table.name, text_table.name)) |
| cursor.drop_table(text_table.name) |
| if self.db_engine is db_connection.IMPALA: |
| with self.cluster.impala.cursor(db_name=self.db_name) as cursor: |
| cursor.invalidate_metadata() |
| cursor.compute_stats() |
| elif self.db_engine is db_connection.HIVE: |
| with self.cluster.hive.cursor(db_name=self.db_name) as cursor: |
| cursor.invalidate_metadata() |
| cursor.compute_stats() |
| else: |
| raise ValueError("db_engine must be of type %s or %s", db_connection.IMPALA, |
| db_connection.HIVE) |
| if postgresql_conn: |
| with postgresql_conn.cursor() as postgresql_cursor: |
| index_tables_in_db_if_possible(postgresql_cursor) |
| |
| def _create_random_table(self, |
| table_name, |
| min_col_count, |
| max_col_count, |
| allowed_storage_formats): |
| '''Create and return a Table with a random number of cols.''' |
| col_count = randint(min_col_count, max_col_count) |
| storage_format = choice(allowed_storage_formats) |
| table = Table(table_name) |
| table.storage_format = storage_format |
| allowed_types = list(TYPES) |
| # Avro doesn't support timestamps yet. |
| if table.storage_format == 'AVRO': |
| allowed_types.remove(Timestamp) |
| # TODO: 'table.cols' returns a copy of all scalar cols, so 'table.cols.append()' |
| # doesn't actually modify the table's columns. 'table.cols' should be changed |
| # to allow access to the real columns. |
| cols = table.cols |
| for col_idx in xrange(col_count): |
| col_type = choice(allowed_types) |
| col_type = choice(filter(lambda type_: issubclass(type_, col_type), EXACT_TYPES)) |
| if issubclass(col_type, VarChar) and not issubclass(col_type, String): |
| col_type = get_varchar_class(randint(1, VarChar.MAX)) |
| elif issubclass(col_type, Char) and not issubclass(col_type, String): |
| col_type = get_char_class(randint(1, Char.MAX)) |
| elif issubclass(col_type, Decimal): |
| max_digits = randint(1, Decimal.MAX_DIGITS) |
| col_type = get_decimal_class(max_digits, randint(1, max_digits)) |
| col = Column( |
| table, |
| '%s_col_%s' % (col_type.__name__.lower(), col_idx + 1), |
| col_type) |
| cols.append(col) |
| table.cols = cols |
| return table |
| |
| def _prepare_table_storage(self, table, db_name): |
| with self.cluster.hive.cursor(db_name=self.db_name) as cursor: |
| cursor.ensure_storage_location(table) |
| hdfs = self.cluster.hdfs.create_client() |
| if hdfs.exists(table.storage_location): |
| hdfs.delete(table.storage_location, recursive=True) |
| hdfs.makedirs(table.storage_location, permission='777') |
| |
| def _run_data_generator_mr_job(self, table_data_generators, db_name): |
| timestamp = int(time()) |
| mapper_input_file = '/tmp/data_gen_%s_mr_input_%s' % (db_name, timestamp) |
| hdfs = self.cluster.hdfs.create_client() |
| if hdfs.exists(mapper_input_file): |
| hdfs.delete(mapper_input_file) |
| reducer_count = 0 |
| mapper_input_data = list() |
| for table_data_generator in table_data_generators: |
| reducer_count += (table_data_generator.row_count / |
| estimate_rows_per_reducer(table_data_generator, MB_PER_REDUCER)) + 1 |
| mapper_input_data.append(serialize(table_data_generator)) |
| hdfs.write(mapper_input_file, data='\n'.join(mapper_input_data)) |
| |
| files = ['common.py', 'db_types.py', 'data_generator_mapred_common.py', |
| 'data_generator_mapper.py', 'data_generator_reducer.py', |
| 'random_val_generator.py'] |
| dir_path = os.path.dirname(__file__) |
| files = [os.path.join(dir_path, f) for f in files] |
| |
| hdfs_output_dir = '/tmp/data_gen_%s_mr_output_%s' % (db_name, timestamp) |
| if hdfs.exists(hdfs_output_dir): |
| hdfs.delete(hdfs_output_dir, recursive=True) |
| |
| LOG.info('Starting MR job to generate data for %s', db_name) |
| self.cluster.yarn.run_mr_job(self.cluster.yarn.find_mr_streaming_jar(), job_args=r''' |
| -D mapred.reduce.tasks=%s \ |
| -D stream.num.map.output.key.fields=2 \ |
| -libjars '%s/share/hadoop/hdfs/lib/*' \ |
| -files %s \ |
| -input %s \ |
| -output %s \ |
| -mapper data_generator_mapper.py \ |
| -reducer data_generator_reducer.py'''.strip() |
| % (reducer_count, os.environ["HADOOP_HOME"], ','.join(files), mapper_input_file, |
| hdfs_output_dir)) |
| |
| |
| if __name__ == '__main__': |
| from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser |
| |
| from tests.comparison import cli_options |
| |
| parser = ArgumentParser( |
| usage='usage: \n' |
| ' %(prog)s [options] [populate]\n\n' |
| ' Create and populate database(s). The Impala database will always be \n' |
| ' included. Postgres is optional. The other databases are not supported.\n\n' |
| ' %(prog)s [options] migrate\n\n' |
| ' Migrate an Impala database to another database type. The destination \n' |
| ' database will be dropped and recreated.', |
| formatter_class=ArgumentDefaultsHelpFormatter) |
| cli_options.add_logging_options(parser) |
| cli_options.add_cluster_options(parser) |
| cli_options.add_db_name_option(parser) |
| cli_options.add_connection_option_groups(parser) |
| |
| group = parser.add_argument_group('Database Population Options') |
| group.add_argument('--randomization-seed', default=1, type=int, |
| help='The randomization will be initialized with this seed. Using the same seed ' |
| 'will produce the same results across runs.') |
| cli_options.add_storage_format_options(group) |
| group.add_argument('--create-data-files', default=False, action='store_true', |
| help='Create files that can be used to repopulate the databases elsewhere.') |
| group.add_argument('--table-count', default=10, type=int, |
| help='The number of tables to generate.') |
| group.add_argument('--min-column-count', default=1, type=int, |
| help='The minimum number of columns to generate per table.') |
| group.add_argument('--max-column-count', default=100, type=int, |
| help='The maximum number of columns to generate per table.') |
| group.add_argument('--min-row-count', default=(10 ** 3), type=int, |
| help='The minimum number of rows to generate per table.') |
| group.add_argument('--max-row-count', default=(10 ** 6), type=int, |
| help='The maximum number of rows to generate per table.') |
| parser.add_argument_group(group) |
| |
| group = parser.add_argument_group('Database Migration Options') |
| group.add_argument('--migrate-table-names', |
| help='Table names should be separated with commas. The default is to migrate all ' |
| 'tables.') |
| parser.add_argument_group(group) |
| parser.add_argument('command', nargs='*', help='The command to run either "populate"' |
| ' or "migrate".') |
| args = parser.parse_args() |
| if len(args.command) > 1: |
| raise Exception('Only one command can be chosen. Requested commands were: %s' |
| % args.command) |
| command = args.command[0] if args.command else 'populate' |
| if command not in ('populate', 'migrate'): |
| raise Exception('Command must either be "populate" or "migrate" but was "%s"' |
| % command) |
| if command == 'migrate' and \ |
| not any((args.use_mysql, args.use_postgresql, args.use_oracle)): |
| raise Exception('At least one destination database must be chosen with ' |
| '--use-<database type>') |
| |
| cli_options.configure_logging(args.log_level, debug_log_file=args.debug_log_file) |
| |
| seed(args.randomization_seed) |
| |
| cluster = cli_options.create_cluster(args) |
| |
| populator = DbPopulator(db_connection.HIVE if args.use_hive else db_connection.IMPALA) |
| if command == 'populate': |
| populator.randomization_seed = args.randomization_seed |
| populator.cluster = cluster |
| populator.db_name = args.db_name |
| populator.min_col_count = args.min_column_count |
| populator.max_col_count = args.max_column_count |
| populator.min_row_count = args.min_row_count |
| populator.max_row_count = args.max_row_count |
| populator.allowed_storage_formats = args.storage_file_formats.split(',') |
| |
| if args.use_hive: |
| with cluster.hive.connect() as conn: |
| with conn.cursor() as cursor: |
| cursor.ensure_empty_db(args.db_name) |
| else: |
| with cluster.impala.connect() as conn: |
| with conn.cursor() as cursor: |
| cursor.invalidate_metadata() |
| cursor.ensure_empty_db(args.db_name) |
| |
| if args.use_postgresql: |
| with cli_options.create_connection(args) as postgresql_conn: |
| with postgresql_conn.cursor() as cursor: |
| cursor.ensure_empty_db(args.db_name) |
| postgresql_conn = cli_options.create_connection(args, db_name=args.db_name) |
| else: |
| postgresql_conn = None |
| populator.populate_db(args.table_count, postgresql_conn=postgresql_conn) |
| else: |
| if args.migrate_table_names: |
| table_names = args.migrate_table_names.split(',') |
| else: |
| table_names = None |
| with cli_options.create_connection(args) as conn: |
| with conn.cursor() as cursor: |
| cursor.ensure_empty_db(args.db_name) |
| with cli_options.create_connection(args, db_name=args.db_name) as conn: |
| with conn.cursor() as dst: |
| with cluster.impala.cursor(db_name=args.db_name) as src: |
| migrate_db(src, dst, include_table_names=table_names) |
