diff --git a/AUTHORS.rst b/AUTHORS.rst index 741c4f9..6bc762c 100644 --- a/AUTHORS.rst +++ b/AUTHORS.rst @@ -6,3 +6,4 @@ Credits * Felipe Hofmann * Kevin Alex Zhang * Carles Sala +* Zhuofan Xie diff --git a/Makefile b/Makefile index 9c8752e..67182ad 100644 --- a/Makefile +++ b/Makefile @@ -87,13 +87,13 @@ install-develop: clean-build clean-pyc ## install the package in editable mode a .PHONY: lint lint: ## check style with flake8 and isort flake8 datatracer tests - isort -c --recursive datatracer tests + isort -c --recursive datatracer tests benchmark .PHONY: fix-lint fix-lint: ## fix lint issues using autoflake, autopep8, and isort - find datatracer tests -name '*.py' | xargs autoflake --in-place --remove-all-unused-imports --remove-unused-variables - autopep8 --in-place --recursive --aggressive datatracer tests - isort --apply --atomic --recursive datatracer tests + find datatracer tests benchmark -name '*.py' | xargs autoflake --in-place --remove-all-unused-imports --remove-unused-variables + autopep8 --in-place --recursive --aggressive datatracer tests benchmark + isort --apply --atomic --recursive datatracer tests benchmark # TEST TARGETS diff --git a/benchmark/README.md b/benchmark/README.md new file mode 100644 index 0000000..d8a3bd2 --- /dev/null +++ b/benchmark/README.md @@ -0,0 +1,55 @@ +# Benchmarking DataTracer +This directory contains code for benchmarking the performance of `DataTracer` +on user-supplied datasets. The datasets for benchmarking can be found in the +`s3://tracer-data` bucket. + +

+ +

+ +Each benchmark - `primary`, `foreign`, and `column` - can be executed by +running the following command + +> datatracer-benchmark --csv /path/to/results.csv + +which will (optionally) generate a CSV file with the benchmark results. + +## Primary Key +Primary key detection is evaluated by: + + - Accuracy. The percent of tables where the primary key was correctly identified. + - Inference time. The amount of time to infer the primary key for all tables. + +We will use leave-one-out validation and report the test performance on each dataset +in the S3 bucket. + +## Foreign Key +Foreign key detection is evaluated by: + + - F1. + - Recall. + - Precision. + - Inference time. + +Note that this assumes that the foreign key primitive returns a set of foreign keys; +in other words, for models that return a score for each candidate foreign key, this +assumes that thresholding is done. + +We will use leave-one-out validation and report the test performance on each dataset +in the S3 bucket. + +## Column Map +Column map detection is evaluated by: + + - F1. + - Recall. + - Precision. + - Inference time. + +Note that this assumes that the column map primitive returns a set of columns that it +thinks contributed to the target derived column. Since each dataset can have multiple +derived columns, this will report a F1/recall/precision/time tuple for each derived +column in the dataset. + +We will use leave-one-out validation and report the test performance on each dataset +in the S3 bucket. diff --git a/benchmark/benchmark.gif b/benchmark/benchmark.gif new file mode 100644 index 0000000..578f0a9 Binary files /dev/null and b/benchmark/benchmark.gif differ diff --git a/benchmark/benchmark.py b/benchmark/benchmark.py new file mode 100644 index 0000000..5400bd2 --- /dev/null +++ b/benchmark/benchmark.py @@ -0,0 +1,189 @@ +import argparse +import os +from io import BytesIO +from time import ctime +from urllib.parse import urljoin +from urllib.request import urlopen +from zipfile import ZipFile + +import boto3 +import pandas as pd + +from column_map_benchmark import benchmark_column_map +from foreign_key_benchmark import benchmark_foreign_key +from how_lineage_benchmark import benchmark_how_lineage +from primary_key_benchmark import benchmark_primary_key + +BUCKET_NAME = 'tracer-data' +DATA_URL = 'http://{}.s3.amazonaws.com/'.format(BUCKET_NAME) + + +def download(data_dir): + """Download benchmark datasets from S3. + + This downloads the benchmark datasets from S3 into the target folder in an + uncompressed format. It skips datasets that have already been downloaded. + + Please make sure an appropriate S3 credential is installed before you call + this method. + + Args: + data_dir: The directory to download the datasets to. + + Returns: + A DataFrame describing the downloaded datasets. + + Raises: + NoCredentialsError: If AWS S3 credentials are not found. + """ + rows = [] + client = boto3.client('s3') + for dataset in client.list_objects(Bucket=BUCKET_NAME)['Contents']: + if not '.zip' in dataset['Key']: + continue + rows.append(dataset) + dataset_name = dataset['Key'].replace(".zip", "") + dataset_path = os.path.join(data_dir, dataset_name) + if os.path.exists(dataset_path): + dataset["Status"] = "Skipped" + print("Skipping %s" % dataset_name) + else: + dataset["Status"] = "Downloaded" + print("Downloading %s" % dataset_name) + with urlopen(urljoin(DATA_URL, dataset['Key'])) as fp: + with ZipFile(BytesIO(fp.read())) as zipfile: + zipfile.extractall(dataset_path) + return pd.DataFrame(rows) + + +def start_with(target, source): + return len(source) <= len(target) and target[:len(source)] == source + + +def aggregate(cmd_name): + cmd_abbrv = {'column': 'ColMap_st', + 'foreign': 'ForeignKey_st', + 'primary': 'PrimaryKey_st' + } + if cmd_name not in cmd_abbrv: + print("Invalid command name!") + return None # invalid command name + cmd_name = cmd_abbrv[cmd_name] + dfs = [] + for file in os.listdir("Reports"): + if start_with(file, cmd_name): + dfs.append(pd.read_csv("Reports/" + file)) + if len(dfs) == 0: + print("No available test results!") + return None + df = pd.concat(dfs, axis=0, ignore_index=True) + os.system("rm Reports/" + cmd_name + "*") # Clean up the caches + return df + + +def _get_parser(): + shared_args = argparse.ArgumentParser(add_help=False) + shared_args.add_argument('--data_dir', type=str, + default=os.path.expanduser("~/tracer_data"), required=False, + help='Path to the benchmark datasets.') + default_csv = "report_" + ctime().replace(" ", "_") + ".csv" + default_csv = default_csv.replace(":", "_") + shared_args.add_argument('--csv', type=str, + default=os.path.expanduser(default_csv), required=False, + help='Path to the CSV file where the report will be written.') + shared_args.add_argument('--ds_name', type=str, + default=None, required=False, + help='Name of the dataset to test on. Default is all available datasets.') + shared_args.add_argument('--problem', type=str, + default=None, required=False, + help='Name of the tests results to aggregate.') + shared_args.add_argument('--primitive', type=str, + default=None, required=False, + help='Name of the primitive to be tested.') + + parser = argparse.ArgumentParser( + prog='datatracer-benchmark', + description='DataTracer Benchmark CLI' + ) + + command = parser.add_subparsers(title='command', help='Command to execute') + parser.set_defaults(benchmark=None) + + subparser = command.add_parser( + 'download', + parents=[shared_args], + help='Download datasets from S3.' + ) + subparser.set_defaults(command=download) + + subparser = command.add_parser( + 'primary', + parents=[shared_args], + help='Primary key benchmark.' + ) + subparser.set_defaults(command=benchmark_primary_key) + + subparser = command.add_parser( + 'foreign', + parents=[shared_args], + help='Foreign key benchmark.' + ) + subparser.set_defaults(command=benchmark_foreign_key) + + subparser = command.add_parser( + 'column', + parents=[shared_args], + help='Column map benchmark.' + ) + subparser.set_defaults(command=benchmark_column_map) + + subparser = command.add_parser( + 'how', + parents=[shared_args], + help='How lineage benchmark.' + ) + subparser.set_defaults(command=benchmark_how_lineage) + + subparser = command.add_parser( + 'aggregate', + parents=[shared_args], + help='Aggregate separate test results.' + ) + subparser.set_defaults(command=aggregate) + + return parser + + +def main(): + parser = _get_parser() + args = parser.parse_args() + if args.command == download: + df = args.command(args.data_dir) + elif args.command == aggregate: + df = args.command(args.problem) + else: + if args.primitive is None: + df = args.command(args.data_dir, args.ds_name) + else: + df = args.command(args.data_dir, args.ds_name, solver=args.primitive) + cmd_abbrv = {'column': 'ColMap_', + 'foreign': 'ForeignKey_', + 'primary': 'PrimaryKey_', + 'how': 'HowLineage_' + } + cmd_str = {benchmark_column_map: 'ColMap_', + benchmark_foreign_key: 'ForeignKey_', + benchmark_primary_key: 'PrimaryKey_', + benchmark_how_lineage: 'HowLineage_', + aggregate: cmd_abbrv[args.problem] if args.problem in cmd_abbrv else '' + } + csv_name = "st_" + args.ds_name + ".csv" if args.ds_name else args.csv + # st is for recognition in the aggregation step + + if csv_name and (args.command in cmd_str) and (df is not None): + df.to_csv("Reports/" + cmd_str[args.command] + csv_name, index=False) + print(df) + + +if __name__ == "__main__": + main() diff --git a/benchmark/column_map_benchmark.py b/benchmark/column_map_benchmark.py new file mode 100644 index 0000000..c733cb4 --- /dev/null +++ b/benchmark/column_map_benchmark.py @@ -0,0 +1,134 @@ +import time +from time import time + +import dask +import pandas as pd +from dask.diagnostics import ProgressBar + +from datatracer import DataTracer, load_datasets + + +@dask.delayed +def evaluate_single_column_map(constraint, tracer, tables): + field = constraint["fields_under_consideration"][0] + related_fields = constraint["related_fields"] + + y_true = set() + for related_field in related_fields: + y_true.add((related_field["table"], related_field["field"])) + + try: + start = time() + ret_dict = tracer.solve(tables, target_table=field["table"], target_field=field["field"]) + y_pred = ret_dict + y_pred = {field for field, score in y_pred.items() if score > 0.0} + end = time() + except BaseException: + return { + "table": field["table"], + "field": field["field"], + "precision": 0, + "recall": 0, + "f1": 0, + "inference_time": 0, + "status": "ERROR", + } + + if len(y_pred) == 0 or len(y_true) == 0 or \ + len(y_true.intersection(y_pred)) == 0: + return { + "table": field["table"], + "field": field["field"], + "precision": 0.0, + "recall": 0.0, + "f1": 0.0, + "inference_time": end - start, + "status": "OK", + } + else: + precision = len(y_true.intersection(y_pred)) / len(y_pred) + recall = len(y_true.intersection(y_pred)) / len(y_true) + f1 = 2.0 * precision * recall / (precision + recall) + + return { + "table": field["table"], + "field": field["field"], + "precision": precision, + "recall": recall, + "f1": f1, + "inference_time": end - start, + "status": "OK", + } + + +@dask.delayed +def column_map(solver, target, datasets): + """Benchmark the column map solver on the target dataset. + + Args: + solver: The name of the column map pipeline. + target: The name of the target dataset. + datases: A dictionary mapping dataset names to (metadata, tables) tuples. + + Returns: + A list of dictionaries mapping metric names to values for each deived column. + """ + datasets = datasets.copy() + metadata, tables = datasets.pop(target) + if not metadata.data.get("constraints"): + return {} # Skip dataset, no constraints found. + + tracer = DataTracer(solver) + tracer.fit(datasets) + + list_of_metrics = [] + for constraint in metadata.data["constraints"]: + list_of_metrics.append(evaluate_single_column_map(constraint, tracer, tables)) + + list_of_metrics = dask.compute(list_of_metrics)[0] + return list_of_metrics + + +def benchmark_column_map(data_dir, dataset_name=None, solver="datatracer.column_map.basic"): + """Benchmark the column map solver. + + This uses leave-one-out validation and evaluates the performance of the + solver on the specified datasets. + + Args: + data_dir: The directory containing the datasets. + dataset_name: The target dataset to test on. If none is provided, will test on all available datasets by default. + solver: The name of the column map pipeline. + + Returns: + A DataFrame containing the benchmark resuls. + """ + datasets = load_datasets(data_dir) + #datasets = sample_datasets(datasets, max_size=20) + dataset_names = list(datasets.keys()) + if dataset_name is not None: + if dataset_name in dataset_names: + dataset_names = [dataset_name] + else: + return None + datasets = dask.delayed(datasets) + dataset_to_metrics = {} + for dataset_name in dataset_names: + dataset_to_metrics[dataset_name] = column_map( + solver=solver, target=dataset_name, datasets=datasets) + + rows = [] + with ProgressBar(): + results = dask.compute(dataset_to_metrics)[0] + for dataset_name, list_of_metrics in results.items(): + for metrics in list_of_metrics: + metrics["dataset"] = dataset_name + rows.append(metrics) + df = pd.DataFrame(rows) + dataset_col = df.pop('dataset') + table_col = df.pop('table') + field_col = df.pop('field') + df.insert(0, 'field', field_col) + df.insert(0, 'table', table_col) + df.insert(0, 'dataset', dataset_col) + return df diff --git a/benchmark/foreign_key_benchmark.py b/benchmark/foreign_key_benchmark.py new file mode 100644 index 0000000..7a0be20 --- /dev/null +++ b/benchmark/foreign_key_benchmark.py @@ -0,0 +1,110 @@ +import time +from time import time + +import dask +import pandas as pd +from dask.diagnostics import ProgressBar + +from datatracer import DataTracer, load_datasets + + +@dask.delayed +def foreign_key(solver, target, datasets): + """Benchmark the foreign key solver on the target dataset. + + Args: + solver: The name of the foreign key pipeline. + target: The name of the target dataset. + datasets: A dictionary mapping dataset names to (metadata, tables) tuples. + + Returns: + A dictionary mapping metric names to values. + """ + datasets = datasets.copy() + metadata, tables = datasets.pop(target) + + tracer = DataTracer(solver) + tracer.fit(datasets) + + y_true = set() + for fk in metadata.get_foreign_keys(): + if not isinstance(fk["field"], str): + continue # Skip composite foreign keys + y_true.add((fk["table"], fk["field"], fk["ref_table"], fk["ref_field"])) + + try: + start = time() + fk_pred = tracer.solve(tables) + end = time() + except BaseException: + return { + "precision": 0, + "recall": 0, + "f1": 0, + "inference_time": 0, + "status": "ERROR" + } + + y_pred = set() + for fk in fk_pred: + y_pred.add((fk["table"], fk["field"], fk["ref_table"], fk["ref_field"])) + + if len(y_pred) == 0 or len(y_true) == 0 or \ + len(y_true.intersection(y_pred)) == 0: + return { + "precision": 0.0, + "recall": 0.0, + "f1": 0.0, + "inference_time": end - start, + "status": "OK" + } + + precision = len(y_true.intersection(y_pred)) / len(y_pred) + recall = len(y_true.intersection(y_pred)) / len(y_true) + f1 = 2.0 * precision * recall / (precision + recall) + + return { + "precision": precision, + "recall": recall, + "f1": f1, + "inference_time": end - start, + "status": "OK" + } + + +def benchmark_foreign_key(data_dir, dataset_name=None, solver="datatracer.foreign_key.standard"): + """Benchmark the foreign key solver. + + This uses leave-one-out validation and evaluates the performance of the + solver on the specified datasets. + + Args: + data_dir: The directory containing the datasets. + dataset_name: The target dataset to test on. If none is provided, will test on all available datasets by default. + solver: The name of the foreign key pipeline. + + Returns: + A DataFrame containing the benchmark resuls. + """ + datasets = load_datasets(data_dir) + #datasets = sample_datasets(datasets, max_size=20) + dataset_names = list(datasets.keys()) + if dataset_name is not None: + if dataset_name in dataset_names: + dataset_names = [dataset_name] + else: + return None + datasets = dask.delayed(datasets) + dataset_to_metrics = {} + for dataset_name in dataset_names: + dataset_to_metrics[dataset_name] = foreign_key( + solver=solver, target=dataset_name, datasets=datasets) + + with ProgressBar(): + results = dask.compute(dataset_to_metrics)[0] + for dataset_name, metrics in results.items(): + metrics["dataset"] = dataset_name + df = pd.DataFrame(list(results.values())) + dataset_col = df.pop('dataset') + df.insert(0, 'dataset', dataset_col) + return df diff --git a/benchmark/how_lineage_benchmark.py b/benchmark/how_lineage_benchmark.py new file mode 100644 index 0000000..e58ccf5 --- /dev/null +++ b/benchmark/how_lineage_benchmark.py @@ -0,0 +1,162 @@ +import time +from time import time + +import dask +import pandas as pd +from dask.diagnostics import ProgressBar + +import datatracer + + +def transform_single_column(tables, column_info): + aggregation = column_info['aggregation'] + column_name = column_info['source_col']['col_name'] + fk = column_info['row_map'] + if aggregation: + transformer = eval(aggregation) + return transformer(tables, fk, column_name) + else: + return tables[column_info['source_col']['table_name']][column_name].fillna(0.0).values + + +def produce_target_column(tables, map_info): + transformation = map_info['transformation'] + if transformation: + transformed_columns = [] + for col_info in map_info['lineage_columns']: + transformed_columns.append(transform_single_column(tables, col_info)) + transformer = eval(transformation) + return transformer(transformed_columns) + else: + return None + + +def approx_equal(num, target, add_margin, multi_margin): + if target >= 0: + return (num <= target * (1 + multi_margin) + add_margin) and (num >= + target * (1 - multi_margin) - add_margin) + else: + return (num <= target * (1 - multi_margin) + add_margin) and (num >= + target * (1 + multi_margin) - add_margin) + + +def approx_equal_arrays(num, target, add_margin, multi_margin): + for n, t in zip(num, target): + if not approx_equal(n, t, add_margin, multi_margin): + return False + return True + + +@dask.delayed +def evaluate_single_lineage(constraint, tracer, tables): + field = constraint["fields_under_consideration"][0] + related_fields = constraint["related_fields"] + + y_true = set() + for related_field in related_fields: + y_true.add((related_field["table"], related_field["field"])) + + try: + start = time() + ret_dict = tracer.solve(tables, target_table=field["table"], target_field=field["field"]) + y_pred = {(col['source_col']['table_name'], col['source_col']['col_name']) + for col in ret_dict['lineage_columns']} + end = time() + except BaseException: + return { + "table": field["table"], + "field": field["field"], + "precision": 0, + "inference_time": 0, + "status": "ERROR", + } + + if len(y_pred) == len(y_true) and \ + len(y_true.intersection(y_pred)) == len(y_pred): + predicted_target = produce_target_column(tables, ret_dict) + target_column = tables[field["table"]][field["field"]].fillna(0.0).values + if approx_equal_arrays(predicted_target, target_column, 1e-8, 1e-8): + precision = 1 + else: + precision = 0 + else: + precision = 0 + return { + "table": field["table"], + "field": field["field"], + "precision": precision, + "inference_time": end - start, + "status": "OK", + } + + +@dask.delayed +def how_lineage(solver, target, datasets): + """Benchmark the how lineage solver on the target dataset. + + Args: + solver: The name of the how lineage pipeline. + target: The name of the target dataset. + datases: A dictionary mapping dataset names to (metadata, tables) tuples. + + Returns: + A list of dictionaries mapping metric names to values for each deived column. + """ + datasets = datasets.copy() + metadata, tables = datasets.pop(target) + if not metadata.data.get("constraints"): + return {} # Skip dataset, no constraints found. + + tracer = datatracer.DataTracer(solver) + tracer.fit(datasets) + + list_of_metrics = [] + for constraint in metadata.data["constraints"]: + list_of_metrics.append(evaluate_single_lineage(constraint, tracer, tables)) + + list_of_metrics = dask.compute(list_of_metrics)[0] + return list_of_metrics + + +def benchmark_how_lineage(data_dir, dataset_name=None, solver="datatracer.how_lineage.basic"): + """Benchmark the how lineage solver. + + This uses leave-one-out validation and evaluates the performance of the + solver on the specified datasets. + + Args: + data_dir: The directory containing the datasets. + dataset_name: The target dataset to test on. If none is provided, will test on all available datasets by default. + solver: The name of the column map pipeline. + + Returns: + A DataFrame containing the benchmark resuls. + """ + datasets = datatracer.load_datasets(data_dir) + dataset_names = list(datasets.keys()) + if dataset_name is not None: + if dataset_name in dataset_names: + dataset_names = [dataset_name] + else: + return None + datasets = dask.delayed(datasets) + dataset_to_metrics = {} + for dataset_name in dataset_names: + dataset_to_metrics[dataset_name] = how_lineage( + solver=solver, target=dataset_name, datasets=datasets) + + rows = [] + with ProgressBar(): + results = dask.compute(dataset_to_metrics)[0] + for dataset_name, list_of_metrics in results.items(): + for metrics in list_of_metrics: + metrics["dataset"] = dataset_name + rows.append(metrics) + df = pd.DataFrame(rows) + dataset_col = df.pop('dataset') + table_col = df.pop('table') + field_col = df.pop('field') + df.insert(0, 'field', field_col) + df.insert(0, 'table', table_col) + df.insert(0, 'dataset', dataset_col) + return df diff --git a/benchmark/primary_key_benchmark.py b/benchmark/primary_key_benchmark.py new file mode 100644 index 0000000..726862e --- /dev/null +++ b/benchmark/primary_key_benchmark.py @@ -0,0 +1,123 @@ +import time +from time import time + +import dask +import pandas as pd +from dask.diagnostics import ProgressBar + +from datatracer import DataTracer, load_datasets + + +@dask.delayed +def primary_key(solver, target, datasets): + """Benchmark the primary key solver on the target dataset. + + Args: + solver: The name of the primary key pipeline. + target: The name of the target dataset. + datases: A dictionary mapping dataset names to (metadata, tables) tuples. + + Returns: + A dictionary mapping metric names to values. + """ + datasets = datasets.copy() + metadata, tables = datasets.pop(target) + + tracer = DataTracer(solver) + tracer.fit(datasets) + + y_true = {} + for table in metadata.get_tables(): + if "primary_key" not in table: + y_true[table["name"]] = set() + elif not isinstance(table["primary_key"], str): + y_true[table["name"]] = set(table["primary_key"]) + else: + y_true[table["name"]] = set([table["primary_key"]]) + + """ + if len(y_true) == 0: + return {} # Skip dataset, no primary keys found. + """ + + correct, total_pred, total_true = 0, 0, 0 + + try: + start = time() + y_pred = tracer.solve(tables) + end = time() + except BaseException: + return { + "precision": 0, + "recall": 0, + "f1": 0, + "inference_time": 0, + "status": "ERROR" + } + for table_name, primary_key in y_true.items(): + ans = y_pred.get(table_name) + if isinstance(ans, str): + ans = set([ans]) + else: + ans = set(ans) + correct += len(ans.intersection(primary_key)) + total_pred += len(ans) + total_true += len(primary_key) + + if correct == 0 or total_pred == 0 or \ + total_true == 0: + return { + "precision": 0.0, + "recall": 0.0, + "f1": 0.0, + "inference_time": end - start, + "status": "OK" + } + precision = correct / total_pred + recall = correct / total_true + f1 = 2 * precision * recall / (precision + recall) + + return { + "precision": precision, + "recall": recall, + "f1": f1, + "inference_time": end - start, + "status": "OK" + } + + +def benchmark_primary_key(data_dir, dataset_name=None, solver="datatracer.primary_key.basic"): + """Benchmark the primary key solver. + + This uses leave-one-out validation and evaluates the performance of the + solver on the specified datasets. + + Args: + data_dir: The directory containing the datasets. + dataset_name: The target dataset to test on. If none is provided, will test on all available datasets by default. + solver: The name of the primary key pipeline. + + Returns: + A DataFrame containing the benchmark resuls. + """ + datasets = load_datasets(data_dir) + dataset_names = list(datasets.keys()) + if dataset_name is not None: + if dataset_name in dataset_names: + dataset_names = [dataset_name] + else: + return None + datasets = dask.delayed(datasets) + dataset_to_metrics = {} + for dataset_name in dataset_names: + dataset_to_metrics[dataset_name] = primary_key( + solver=solver, target=dataset_name, datasets=datasets) + + with ProgressBar(): + results = dask.compute(dataset_to_metrics)[0] + for dataset_name, metrics in results.items(): + metrics["dataset"] = dataset_name + df = pd.DataFrame(list(results.values())) + dataset_col = df.pop('dataset') + df.insert(0, 'dataset', dataset_col) + return df diff --git a/datatracer/__init__.py b/datatracer/__init__.py index 9ccfd89..3f18d2f 100644 --- a/datatracer/__init__.py +++ b/datatracer/__init__.py @@ -12,6 +12,7 @@ from datatracer.core import PRETRAINED_DIR, DataTracer from datatracer.data import get_demo_data, load_dataset, load_datasets +from datatracer.data_sampler import sample_datasets _BASE_PATH = os.path.abspath(os.path.dirname(__file__)) _JSONS_PATH = os.path.join(_BASE_PATH, 'jsons') @@ -26,6 +27,8 @@ 'get_primitives', 'load_dataset', 'load_datasets', + 'sample_datasets', + 'how_lineage' ) diff --git a/datatracer/column_map/base.py b/datatracer/column_map/base.py index 435bbf3..a257ce5 100644 --- a/datatracer/column_map/base.py +++ b/datatracer/column_map/base.py @@ -4,15 +4,15 @@ class ColumnMapSolver: """Base Solver for the data lineage problem of column dependency.""" - def fit(self, list_of_databases): + def fit(self, dict_of_databases): """Fit this solver. Args: - list_of_databases (list): - List of tuples containing ``MetaData`` instnces and table dictinaries, - which contain table names as input and ``pandas.DataFrames`` as values. + dict_of_databases (dict): + Map from database names to tuples containing ``MetaData`` + instances and table dictionaries, which contain table names + as input and ``pandas.DataFrames`` as values. """ - pass def solve(self, tables, foreign_keys, target_table, target_field): """Find the fields which contributed to the target_field the most. diff --git a/datatracer/column_map/basic.py b/datatracer/column_map/basic.py index 1c0e48f..ac53871 100644 --- a/datatracer/column_map/basic.py +++ b/datatracer/column_map/basic.py @@ -1,5 +1,8 @@ import logging +import time +from itertools import combinations +import numpy as np from sklearn.ensemble import RandomForestRegressor from datatracer.column_map.base import ColumnMapSolver @@ -8,12 +11,97 @@ LOGGER = logging.getLogger(__name__) +def approx_equal(num, target, add_margin, multi_margin): + if target >= 0: + return (num <= target * (1 + multi_margin) + add_margin) and\ + (num >= target * (1 - multi_margin) - add_margin) + else: + return (num <= target * (1 - multi_margin) + add_margin) and\ + (num >= target * (1 + multi_margin) - add_margin) + + +def approx_equal_arrays(num, target, add_margin, multi_margin): + for n, t in zip(num, target): + if not approx_equal(n, t, add_margin, multi_margin): + return False + return True + + +def check_sum(indicies, X, y, add_margin, multi_margin): + return approx_equal_arrays(X[:, indicies].sum(axis=1), y, add_margin, multi_margin) + + +def check_avg(indicies, X, y, add_margin, multi_margin): + return approx_equal_arrays(X[:, indicies].sum( + axis=1) / len(indicies), y, add_margin, multi_margin) + + +def check_diff(indicies, X, y, add_margin, multi_margin): + pred_y = X[:, indicies[0]] - X[:, indicies[1]] + return approx_equal_arrays(pred_y, y, 0, 0) + + +def detect_restricted_reg(X, y, add_margin=1e-4, mult_margin=1e-4, max_feature=5, timeout=3600): + """ + This method runs a restricted regression where the target column is either the sum + or difference of several columns in the given table, or the average of several columns + in the given table. + + Returns: + (str, tuple): a string ("sum", "diff", "avg" or "None") representing the operation, + and a tuple of coeffs. + """ + start_time = time.time() + + dot_prods = (X.T).dot(y) + length = len(dot_prods) + y2 = y.dot(y) + for num_feature in range(1, max_feature + 1): + for combo in combinations(range(length), num_feature): + if time.time() - start_time > timeout: + return "None", None + + indicies = list(combo) + if approx_equal(dot_prods[indicies].sum(), y2, add_margin, mult_margin): + if check_sum(indicies, X, y, add_margin, mult_margin): + weights = [0] * length + for ind in indicies: + weights[ind] = 1 + return "sum", weights + if (num_feature > 1) and approx_equal( + dot_prods[indicies].sum() / num_feature, y2, add_margin, mult_margin): + if check_avg(indicies, X, y, add_margin, mult_margin): + weights = [0] * length + for ind in indicies: + weights[ind] = 1.0 / num_feature + return "avg", weights + if num_feature == 2: + if approx_equal(dot_prods[indicies[0]] - dot_prods[indicies[1]], + y2, add_margin, mult_margin): + if check_diff(indicies, X, y, add_margin, mult_margin): + weights = [0] * length + weights[indicies[0]] = 1 + weights[indicies[1]] = -1 + return "diff", weights + if approx_equal(dot_prods[indicies[1]] - dot_prods[indicies[0]], + y2, add_margin, mult_margin): + if check_diff(indicies[::-1], X, y, add_margin, mult_margin): + weights = [0] * length + weights[indicies[0]] = -1 + weights[indicies[1]] = 1 + return "diff", weights + return "None", None + + class BasicColumnMapSolver(ColumnMapSolver): """Basic Solver for the data lineage problem of column dependency.""" - def __init__(self, *args, **kwargs): + def __init__(self, threshold=0.1, *args, **kwargs): self._model_args = args self._model_kwargs = kwargs + self._threshold = threshold + self._linear_weight_threshold = 1e-4 + self._linear_score_threshold = 0.95 def _get_importances(self, X, y): model = RandomForestRegressor(*self._model_args, **self._model_kwargs) @@ -21,6 +109,12 @@ def _get_importances(self, X, y): return model.feature_importances_ + def _convert_linear_importances(self, weights): + new_weights = (weights > self._linear_weight_threshold) / \ + sum(weights > self._linear_weight_threshold) + + return new_weights + def solve(self, tables, foreign_keys, target_table, target_field): """Find the fields which contributed to the target_field the most. @@ -46,6 +140,29 @@ def solve(self, tables, foreign_keys, target_table, target_field): transformer = Transformer(tables, foreign_keys) X, y = transformer.forward(target_table, target_field) + if len(X.shape) != 2: # invalid X shape + return {} + elif X.shape[0] == 0 or X.shape[1] == 0: # empty dimension + return {} + + try: + restricted_linear_type, weights = detect_restricted_reg(X, y) + if restricted_linear_type != "None": + importances = self._convert_linear_importances(np.array(weights)) + else: + importances = self._get_importances(X, y) + except BaseException: + importances = self._get_importances(X, y) - importances = self._get_importances(X, y) - return transformer.backward(importances) + ret_dict = transformer.backward(importances) + flag = True + while flag: + flag = False + new_rets = ret_dict.copy() + total_score = sum(ret_dict.values()) + for field, score in ret_dict.items(): + if score < total_score * self._threshold / len(ret_dict): + del new_rets[field] + flag = True + ret_dict = new_rets + return ret_dict diff --git a/datatracer/column_map/transformer.py b/datatracer/column_map/transformer.py index 28b7441..856bf62 100644 --- a/datatracer/column_map/transformer.py +++ b/datatracer/column_map/transformer.py @@ -116,6 +116,9 @@ def backward(self, feature_importances): """ obj = {} for column, importance in zip(self.columns, feature_importances): - obj[column] = importance + if column in obj: + obj[column] += importance + else: + obj[column] = importance return obj diff --git a/datatracer/core.py b/datatracer/core.py index 349ca7e..a51a0c7 100644 --- a/datatracer/core.py +++ b/datatracer/core.py @@ -12,6 +12,8 @@ from mlblocks import MLPipeline PRETRAINED_DIR = os.path.join(os.path.dirname(__file__), 'pretrained') +PIPELINE_DIR = os.path.join(os.path.dirname(__file__), 'jsons/pipelines') +PRIMITIVE_DIR = os.path.join(os.path.dirname(__file__), 'jsons/primitives') class DataTracer: @@ -33,9 +35,27 @@ class DataTracer: def _get_mlpipeline(self): pipeline = self._pipeline - if isinstance(pipeline, str) and os.path.isfile(pipeline): - with open(pipeline) as json_file: - pipeline = json.load(json_file) + if isinstance(pipeline, str): + if os.path.isfile(pipeline): + with open(pipeline) as json_file: + pipeline = json.load(json_file) + elif os.path.isfile(os.path.join(PIPELINE_DIR, pipeline + '.json')): + with open(os.path.join(PIPELINE_DIR, pipeline + '.json')) as json_file: + pipeline = json.load(json_file) + + if isinstance(pipeline, dict): + if 'primitives' in pipeline: + for idx in range(len(pipeline['primitives'])): + primitive = pipeline['primitives'][idx] + if isinstance(primitive, str): + if os.path.isfile(primitive): + with open(primitive) as json_file: + primitive = json.load(json_file) + elif os.path.isfile(os.path.join(PRIMITIVE_DIR, primitive + '.json')): + with open(os.path.join(PRIMITIVE_DIR, primitive + '.json'))\ + as json_file: + primitive = json.load(json_file) + pipeline['primitives'][idx] = primitive mlpipeline = MLPipeline(pipeline) if self._hyperparameters: @@ -52,15 +72,13 @@ def fit(self, datasets): """Fit the pipeline to the given data. Args: - datasets (list or dict): - List or dict of tuples containing a MetaData instance and a dict - with the tables of the dataset loaded as DataFrames. + datasets (dict): + Dict mapping dataset names to tuples containing a MetaData + instance and a dict with the tables of the dataset loaded + as DataFrames. """ - if isinstance(datasets, dict): - datasets = list(datasets.values()) - self._mlpipeline = self._get_mlpipeline() - self._mlpipeline.fit(list_of_databases=datasets, tables={}) + self._mlpipeline.fit(dict_of_databases=datasets, tables={}) def solve(self, tables=None, **kwargs): """Solve the data lineage problem. diff --git a/datatracer/data_sampler.py b/datatracer/data_sampler.py new file mode 100644 index 0000000..5daea8b --- /dev/null +++ b/datatracer/data_sampler.py @@ -0,0 +1,244 @@ +# -*- coding: utf-8 -*- + +"""DataTracer core module. + +This module introduces tools for sampling from databases while respecting the row lineage. +""" + +import random + +import dask +from dask.diagnostics import ProgressBar + + +def calculate_size(transformed_dataset): + """Helper function to calculate the total size of a dataset + + Args: + transformed_dataset (dict): a ``TransformedDataset`` instance, which maps (str) table name + to {'size': (float) size of the table in byte, 'row_size': (float) the size of a + row in byte, 'entries': (set) the column names, 'chosen': (set) the rows selected} + + Returns: + float: the dataset size in byte + """ + size = 0 + for table in transformed_dataset.values(): + size += table['row_size'] * len(table['chosen']) + return size + + +def transform_dataset(metadata, dataset): + """Pack the foreign key relations, sizes, and the rows selected of a dataset into dictionaries. + + Args: + metadata (dict): a ``MetaData`` instance + dataset (dict): maps table name to pd.DataFrame object + + Returns: + dict: a ``TransformedForeignKey`` instance, which maps (str) table name to (list(tuple)) + its associated foreign key relations + dict: a ``TransformedDataset`` instance + float: the dataset size in byte + """ + fks = metadata.get_foreign_keys() + transformed_fk = {} + key_columns = {table_name: set() for table_name in dataset} + for fk in fks: + table, all_field, ref_table, all_ref_field =\ + fk["table"], fk["field"], fk["ref_table"], fk["ref_field"] + if isinstance(all_field, str): + all_field = [all_field] + all_ref_field = [all_ref_field] + for field, ref_field in zip(all_field, all_ref_field): + key_columns[table].add(field) + key_columns[ref_table].add(ref_field) + if ref_table not in transformed_fk: + transformed_fk[ref_table] = [] + transformed_fk[ref_table].append((ref_table, ref_field, table, field)) + transformed_dataset = {} + size = 0 + for table_name in dataset: + table = dataset[table_name] + columns = key_columns[table_name] + transformed_table = {'size': table.memory_usage().sum(), + 'row_size': float(table.memory_usage().sum()) / len(table), + 'entries': {col: {} for col in columns}, + 'chosen': set(range(len(table))), } + for idx in range(len(table)): + for col in columns: + val = table.iloc[idx][col] + if val not in transformed_table['entries'][col]: + transformed_table['entries'][col][val] = [] + transformed_table['entries'][col][val].append(idx) + transformed_dataset[table_name] = transformed_table + size += transformed_table['size'] + return transformed_fk, transformed_dataset, size + + +def backward_transform(transformed_dataset, dataset): + """Transform a ``TransformedDataset`` instance back into a dictionary mapping name + to pd.DataFrame objects + + Args: + transformed_dataset (dict): a ``TransformedDataset`` instance + dataset (dict): a dictionary mapping table name to pd.DataFrame object + + Returns: + dict: a dictionary mapping table name to pd.DataFrame object + """ + new_dataset = {} + for table_name in dataset: + idxes = list(transformed_dataset[table_name]['chosen']) + new_dataset[table_name] = dataset[table_name].iloc[idxes] + return new_dataset + + +def remove_row(dataset, transformed_fk, transformed_dataset, table_name, idx): + """Remove a row from a table, and recursively removed all other rows associated + by some foreign key relations + + Args: + dataset (dict): a dictionary mapping table name to pd.DataFrame object + transformed_fk (dict): a ``TransformedForeignKey`` instance + transformed_dataset (dict): a ``TransformedDataset`` instance + table_name (string): the name of the table in which a row is to be removed + idx (int): the index of the row to be removed + + Returns: + None if at least of the tables is empty after the recursive removal. True otherwise. + """ + if idx in transformed_dataset[table_name]['chosen']: + transformed_dataset[table_name]['chosen'].remove(idx) + if len(transformed_dataset[table_name]['chosen']) == 0: + return None + row = dataset[table_name].iloc[idx] + if table_name in transformed_fk: + for table, col, other_table, other_col in transformed_fk[table_name]: + val = row[col] + if val in transformed_dataset[other_table]['entries'][other_col]: + for new_idx in transformed_dataset[other_table]['entries'][other_col][val]: + if new_idx in transformed_dataset[other_table]['chosen']: + if remove_row(dataset, transformed_fk, transformed_dataset, + other_table, new_idx) is None: + return None + return True + + +def get_root_tables(metadata): + """Get all root tables (tables who are never a child table in a foreign key relation) + of the dataset. + + Args: + metadata (dict): a ``MetaData`` instance + + Returns: + set: the root table names + """ + all_tables = {table['name'] for table in metadata.get_tables()} + + for fk in metadata.get_foreign_keys(): + if fk['table'] in all_tables: + all_tables.remove(fk['table']) + if len(all_tables) > 0: + return all_tables + else: + return {table['name'] for table in metadata.get_tables()} + + +@dask.delayed +def sample_dataset(metadata=None, dataset=None, max_size=None, max_ratio=1.0, drop_threshold=None, + rand_seed=0, database_name=None, dict_of_databases=None): + """Sample from a dataset + + Args: + metadata (dict): a ``MetaData`` instance. + Will be extracted from datasets if None is provided. + dataset (dict): maps table name to pd.DataFrame object. + Will be extracted from datasets if None is provided. + max_size (float): the target size to sample down to, in MB. + max_ratio (float): the target fraction to sample down to. + drop_threshold (float): the maximum dataset size allowed. + rand_seed (int): seed for random. + database_name (str): name of the dataset. + dict_of_databases (dict): maps (str) dataset name to (metadata, dataset) tuple. + + Returns: + tuple (None, str) describing the reason of dropping the dataset if it is dropped. + dict, which maps table name to pd.DataFrame object, otherwise. + """ + if dict_of_databases is not None: + metadata, dataset = dict_of_databases[database_name] + + if len(dataset) == 0: + return (None, 'empty') # empty dataset + + size = sum([table.memory_usage().sum() for _, table in dataset.items()]) + if max_size is not None: + max_size *= (1024.0**2) # input max_size is in MB + else: + max_size = size + target_size = min(max_size, size * float(max_ratio)) + + if drop_threshold is None: + drop_threshold = 5 * target_size + if size <= target_size + 1: + return dataset + elif size > drop_threshold: + return (None, 'size') + + random.seed(rand_seed) + transformed_fk, transformed_dataset, size = transform_dataset(metadata, dataset) + root_tables = get_root_tables(metadata) + while calculate_size(transformed_dataset) > target_size + \ + 1: # +1 is for preventing precision issues + table_name = random.sample(root_tables, 1)[0] + if len(transformed_dataset[table_name]['chosen']) > 0: + idx = random.sample(transformed_dataset[table_name]['chosen'], 1)[0] + if remove_row(dataset, transformed_fk, transformed_dataset, table_name, idx) is None: + return (None, 'empty') + else: + return (None, 'empty') + + return backward_transform(transformed_dataset, dataset) + + +def sample_datasets(dict_of_databases, max_size=None, max_ratio=1.0, + drop_threshold=None, rand_seed=0): + """Sample from multiple datasets + + Args: + dict_of_databases (dict): maps (str) dataset name to (metadata, dataset) tuple. + max_size (float): the target size to sample down to, in MB. + max_ratio (float): the target fraction to sample down to. + drop_threshold (float): the maximum dataset size allowed. + rand_seed (int): seed for random. + + Returns: + dict: maps (str) dataset name to (metadata, dataset) tuple. + """ + db_names = list(dict_of_databases.keys()) + immediate_dict_of_db = dict_of_databases + dict_of_databases = dask.delayed(dict_of_databases) + new_dict_of_databases = {} + for database_name in db_names: + new_dict_of_databases[database_name] = sample_dataset( + max_size=max_size, + max_ratio=max_ratio, + drop_threshold=drop_threshold, + rand_seed=rand_seed, + dict_of_databases=dict_of_databases, + database_name=database_name) + with ProgressBar(): + new_dict_of_databases = dask.compute(new_dict_of_databases)[0] + for database_name in db_names: + if isinstance(new_dict_of_databases[database_name], tuple): + if new_dict_of_databases[database_name][1] == 'empty': + print("%s is dropped because of empty tables when sampling" % database_name) + elif new_dict_of_databases[database_name][1] == 'size': + print("%s is dropped because it's too big" % database_name) + del new_dict_of_databases[database_name] + else: + metadata = immediate_dict_of_db[database_name][0] + new_dict_of_databases[database_name] = (metadata, new_dict_of_databases[database_name]) + return new_dict_of_databases diff --git a/datatracer/foreign_key/base.py b/datatracer/foreign_key/base.py index d31a96e..82a336e 100644 --- a/datatracer/foreign_key/base.py +++ b/datatracer/foreign_key/base.py @@ -3,15 +3,15 @@ class ForeignKeySolver(): - def fit(self, list_of_databases): + def fit(self, dict_of_databases): """Fit this solver. Args: - list_of_databases (list): - List of tuples containing ``MetaData`` instnces and table dictinaries, - which contain table names as input and ``pandas.DataFrames`` as values. + dict_of_databases (dict): + Map from database names to tuples containing ``MetaData`` + instances and table dictionaries, which contain table names + as input and ``pandas.DataFrames`` as values. """ - pass def solve(self, tables, primary_keys=None): """Solve the foreign key detection problem. diff --git a/datatracer/foreign_key/standard.py b/datatracer/foreign_key/standard.py index 1210d59..fd027e1 100644 --- a/datatracer/foreign_key/standard.py +++ b/datatracer/foreign_key/standard.py @@ -9,7 +9,7 @@ class StandardForeignKeySolver(ForeignKeySolver): - def __init__(self, threshold=0.9, add_details=False, *args, **kwargs): + def __init__(self, threshold=[i / 20 for i in range(20)], add_details=False, *args, **kwargs): self._threshold = threshold self._add_details = add_details self._model_args = args @@ -50,24 +50,41 @@ def _feature_vector(self, parent_col, child_col): 1.0 if child_col.dtype == "object" else 0.0, ] - def fit(self, list_of_databases): + def fit(self, dict_of_databases): """Fit this solver. Args: - list_of_databases (list): - List of tuples containing ``MetaData`` instnces and table dictinaries, - which contain table names as input and ``pandas.DataFrames`` as values. + dict_of_databases (dict): + Map from database names to tuples containing ``MetaData`` + instances and table dictionaries, which contain table names + as input and ``pandas.DataFrames`` as values. """ X, y = [], [] - for metadata, tables in tqdm(list_of_databases, "extracting features"): - fks = set() - for fk in metadata.get_foreign_keys(): - if isinstance(fk["field"], str): - fks.add((fk["table"], fk["field"], fk["ref_table"], fk["ref_field"])) + iterator = tqdm(dict_of_databases.items()) + for database_name, (metadata, tables) in iterator: + iterator.set_description("Extracting features from %s" % database_name) + fks = metadata.get_foreign_keys() + tables_info = {table_info['name']: table_info for table_info in metadata.get_tables()} + fks_new = [] + for fk in fks: + if isinstance(fk["field"], list): + for field, ref_field in zip(fk["field"], fk["ref_field"]): + fks_new.append((fk["table"], field, fk["ref_table"], ref_field)) + else: + fks_new.append((fk["table"], fk["field"], fk["ref_table"], fk["ref_field"])) + fks = set(fks_new) for t1, t2 in permutations(tables.keys(), r=2): - for c1 in tables[t1].columns: - for c2 in tables[t2].columns: + table = tables_info[t2] + if "primary_key" not in table: + t2_columns = tables[t2].columns + elif not isinstance(table["primary_key"], str): + t2_columns = table["primary_key"] + else: + t2_columns = [table["primary_key"]] + + for c2 in t2_columns: + for c1 in tables[t1].columns: if tables[t1][c1].dtype.kind != tables[t2][c2].dtype.kind: continue @@ -81,6 +98,26 @@ def fit(self, list_of_databases): self.model = RandomForestClassifier(*self._model_args, **self._model_kwargs) self.model.fit(X, y) + if isinstance(self._threshold, list): + best_f1 = -float('inf') + best_threshold = None + pred_y = self.model.predict(X) + len_true = sum(y) + for threshold in self._threshold: + filtered_y = (pred_y >= threshold).astype(float) + intersect = sum(filtered_y * y) + len_pred = sum(filtered_y) + if intersect * len_true * len_pred == 0: + f1 = 0 + else: + precision = intersect / len_pred + recall = intersect / len_true + f1 = 2.0 * precision * recall / (precision + recall) + if f1 > best_f1: + best_f1 = f1 + best_threshold = threshold + self._threshold = best_threshold + def solve(self, tables, primary_keys=None): """Solve the foreign key detection problem. @@ -101,7 +138,13 @@ def solve(self, tables, primary_keys=None): X, foreign_keys = [], [] for t1, t2 in permutations(tables.keys(), r=2): for c1 in tables[t1].columns: - for c2 in tables[t2].columns: + if (primary_keys is None) or (t2 not in primary_keys): + t2_columns = tables[t2].columns + elif not isinstance(primary_keys[t2], str): + t2_columns = primary_keys[t2] + else: + t2_columns = [primary_keys[t2]] + for c2 in t2_columns: if tables[t1][c1].dtype.kind != tables[t2][c2].dtype.kind: continue diff --git a/datatracer/how_lineage/__init__.py b/datatracer/how_lineage/__init__.py new file mode 100644 index 0000000..14fcec2 --- /dev/null +++ b/datatracer/how_lineage/__init__.py @@ -0,0 +1,18 @@ +from datatracer.how_lineage.base import HowLineageSolver +from datatracer.how_lineage.basic import BasicHowLineageSolver +from datatracer.how_lineage.table_transforms import ( + columns_avg, columns_diff, columns_sum, entries_count, entries_max, entries_min, entries_std, + entries_sum) + +__all__ = ( + 'HowLineageSolver', + 'BasicHowLineageSolver', + 'entries_count', + 'entries_sum', + 'entries_min', + 'entries_max', + 'entries_std', + 'columns_sum', + 'columns_diff', + 'columns_avg' +) diff --git a/datatracer/how_lineage/base.py b/datatracer/how_lineage/base.py new file mode 100644 index 0000000..d98f65f --- /dev/null +++ b/datatracer/how_lineage/base.py @@ -0,0 +1,39 @@ +"""Column Mapping base class.""" + + +class HowLineageSolver: + """Base Solver for the data lineage problem of how lineage.""" + + def fit(self, dict_of_databases): + """Fit this solver. + + Args: + dict_of_databases (dict): + Map from database names to tuples containing ``MetaData`` + instances and table dictionaries, which contain table names + as input and ``pandas.DataFrames`` as values. + """ + + def solve(self, tables, foreign_keys, target_table, target_field): + """Find the fields which contributed to the target_field the most. + + The output is a dictionary containing the fields that contributed the + most to the given target field as keys, specified as a tuple containing + both table name and field name, and the score obtained as values. + + Args: + tables (dict): + Dict containing table names as input and ``pandas.DataFrames`` + as values. + foreign_keys (list): + List of foreign key specifications. + target_table (str): + Name of the table that contains the target field. + target_field (str): + Name of the target field. + + Returns: + dict: + Dictionary of field specification tuples and scores. + """ + raise NotImplementedError() diff --git a/datatracer/how_lineage/basic.py b/datatracer/how_lineage/basic.py new file mode 100644 index 0000000..56d2969 --- /dev/null +++ b/datatracer/how_lineage/basic.py @@ -0,0 +1,161 @@ +import logging +import time +from itertools import combinations + +from sklearn.ensemble import RandomForestRegressor + +from datatracer.how_lineage.base import HowLineageSolver +from datatracer.how_lineage.transformer import Transformer + +LOGGER = logging.getLogger(__name__) + + +def approx_equal(num, target, add_margin, multi_margin): + if target >= 0: + return (num <= target * (1 + multi_margin) + add_margin) and\ + (num >= target * (1 - multi_margin) - add_margin) + else: + return (num <= target * (1 - multi_margin) + add_margin) and\ + (num >= target * (1 + multi_margin) - add_margin) + + +def approx_equal_arrays(num, target, add_margin, multi_margin): + for n, t in zip(num, target): + if not approx_equal(n, t, add_margin, multi_margin): + return False + return True + + +def check_sum(indicies, X, y, add_margin, multi_margin): + return approx_equal_arrays(X[:, indicies].sum(axis=1), y, add_margin, multi_margin) + + +def check_avg(indicies, X, y, add_margin, multi_margin): + return approx_equal_arrays(X[:, indicies].sum( + axis=1) / len(indicies), y, add_margin, multi_margin) + + +def check_diff(indicies, X, y, add_margin, multi_margin): + pred_y = X[:, indicies[0]] - X[:, indicies[1]] + return approx_equal_arrays(pred_y, y, 0, 0) + + +def detect_restricted_reg(X, y, add_margin=1e-4, mult_margin=1e-4, max_feature=5, timeout=3600): + """ + This method runs a restricted regression where the target column is either the sum + or difference of several columns in the given table, or the average of several columns + in the given table. + + Returns: + (str, tuple): a string ("sum", "diff", "avg" or "None") representing the operation, + and a tuple of coeffs. + """ + start_time = time.time() + + dot_prods = (X.T).dot(y) + length = len(dot_prods) + y2 = y.dot(y) + for num_feature in range(1, max_feature + 1): + for combo in combinations(range(length), num_feature): + if time.time() - start_time > timeout: + return "None", None + + indicies = list(combo) + if approx_equal(dot_prods[indicies].sum(), y2, add_margin, mult_margin): + if check_sum(indicies, X, y, add_margin, mult_margin): + return "sum", indicies + if (num_feature > 1) and approx_equal( + dot_prods[indicies].sum() / num_feature, y2, add_margin, mult_margin): + if check_avg(indicies, X, y, add_margin, mult_margin): + return "avg", indicies + if num_feature == 2: + if approx_equal(dot_prods[indicies[0]] - dot_prods[indicies[1]], + y2, add_margin, mult_margin): + if check_diff(indicies, X, y, add_margin, mult_margin): + return "diff", indicies + if approx_equal(dot_prods[indicies[1]] - dot_prods[indicies[0]], + y2, add_margin, mult_margin): + if check_diff(indicies[::-1], X, y, add_margin, mult_margin): + return "diff", indicies[::-1] + return "None", None + + +class BasicHowLineageSolver(HowLineageSolver): + """Basic Solver for the data lineage problem of how lineage.""" + + def __init__(self, threshold=0.1, *args, **kwargs): + self._model_args = args + self._model_kwargs = kwargs + self._threshold = threshold + self._linear_weight_threshold = 1e-4 + self._linear_score_threshold = 0.95 + + def _get_importances(self, X, y): + model = RandomForestRegressor(*self._model_args, **self._model_kwargs) + model.fit(X, y) + + return model.feature_importances_ + + def _convert_linear_importances(self, weights): + new_weights = (weights > self._linear_weight_threshold) / \ + sum(weights > self._linear_weight_threshold) + + return new_weights + + def solve(self, tables, foreign_keys, target_table, target_field): + """Find the fields which contributed to the target_field the most. + + The output is a dictionary containing the fields that contributed the + most to the given target field as keys, specified as a tuple containing + both table name and field name, and the score obtained as values. + + Args: + tables (dict): + Dict containing table names as input and ``pandas.DataFrames`` + as values. + foreign_keys (list): + List of foreign key specifications. + target_table (str): + Name of the table that contains the target field. + target_field (str): + Name of the target field. + + Returns: + dict: + Dictionary of field specification tuples and scores. + """ + transformer = Transformer(tables, foreign_keys) + + X, y = transformer.forward(target_table, target_field) + if len(X.shape) != 2: # invalid X shape + print("Encountered invalid X shape in how-lineage detection.\ + Please check if any table is empty or if foreign keys have been provided.") + return {"lineage_columns": [], + "transformation": ""} + elif X.shape[0] == 0 or X.shape[1] == 0: # empty dimension + print("Encountered invalid X shape in how-lineage detection.\ + Please check if any table is empty or if foreign keys have been provided.") + return {"lineage_columns": [], + "transformation": ""} + + try: + restricted_linear_type, indicies = detect_restricted_reg(X, y) + if restricted_linear_type == "None": + print("Failed to detect any basic linear maps in how-lineage detection.") + return {"lineage_columns": [], + "transformation": ""} + except BaseException as e: + print("Encountered an error in how-lineage detection, though very likely a\ + standard timeout. Error message is as below.") + print(e) + return {"lineage_columns": [], + "transformation": ""} + + lineage = [transformer.columns[idx] for idx in indicies] + + linear_map_dict = {"sum": "datatracer.how_lineage.columns_sum", + "diff": "datatracer.how_lineage.columns_diff", + "avg": "datatracer.how_lineage.columns_avg"} + + return {"lineage_columns": lineage, + "transformation": linear_map_dict[restricted_linear_type]} diff --git a/datatracer/how_lineage/table_transforms.py b/datatracer/how_lineage/table_transforms.py new file mode 100644 index 0000000..bdfe366 --- /dev/null +++ b/datatracer/how_lineage/table_transforms.py @@ -0,0 +1,43 @@ +def transform(tables, fk, col_name, op): + child_table = tables[fk["table"]].copy() + parent_table = tables[fk["ref_table"]].copy() + child_table["_dummy_"] = 0.0 + if len(child_table.columns) <= 1: + raise ValueError("Invalid lineage table/transformation combo!") + result_col = op(child_table.groupby(fk["field"]))[col_name] + parent_table = parent_table.set_index(fk["ref_field"]) + parent_table = parent_table.join(result_col).reset_index() + + return parent_table[col_name].fillna(0.0).values + + +def entries_count(tables, fk, col_name): + return transform(tables, fk, '_dummy_', lambda x: x.count()) + + +def entries_sum(tables, fk, col_name): + return transform(tables, fk, col_name, lambda x: x.sum()) + + +def entries_min(tables, fk, col_name): + return transform(tables, fk, col_name, lambda x: x.min()) + + +def entries_max(tables, fk, col_name): + return transform(tables, fk, col_name, lambda x: x.max()) + + +def entries_std(tables, fk, col_name): + return transform(tables, fk, col_name, lambda x: x.std()) + + +def columns_sum(columns): + return sum(columns) + + +def columns_diff(columns): + return columns[0] - columns[1] + + +def columns_avg(columns): + return sum(columns) / len(columns) diff --git a/datatracer/how_lineage/transformer.py b/datatracer/how_lineage/transformer.py new file mode 100644 index 0000000..af2d272 --- /dev/null +++ b/datatracer/how_lineage/transformer.py @@ -0,0 +1,128 @@ +import numpy as np + + +class Transformer: + + def __init__(self, tables, foreign_keys): + """ + The `Transformer` class provides an interface between the database and + the column mapping solver. + + For example, during the forwards pass, it could take a date field and + transform into three columns - day, month, and year - which will allow + the solver to potentially identify the lineage of some target column. + + Then, during the backwards pass, it can take the scores produced by the + solver (i.e. the solver assigns scores indicating how important the day, + month, and year columns are to predicting the target column) and combine + them into a single score for the `date` field. + """ + self.tables = tables + self.foreign_keys = foreign_keys + + def forward(self, table, field): + """ + This function returns a (X, y) tuple containing numerical values which + is suitable for machine learning libraries. The `X` array contains + data from columns that are potentially related to the target field. The + `y` array contains the values of the target field. + """ + df = self.tables[table] + df = df.select_dtypes("number") + df = df.fillna(0.0) + X, y = df.drop([field], axis=1), df[field] + self.columns = [{"source_col": {"table_name": table, "col_name": col_name}, + "row_map": {}, + "aggregation": "" + } for col_name in X.columns] + X, y = X.values, y.values + + X_new, columns_new = self._get_counts(table) + if columns_new: + X = np.concatenate([X, X_new], axis=1) + self.columns.extend(columns_new) + + X_new, columns_new = self._get_aggregations(table) + if columns_new: + X = np.concatenate([X, X_new], axis=1) + self.columns.extend(columns_new) + + return X, y + + def _get_counts(self, table): + """ + Get the foreign keys where the given table is the parent. + """ + X, columns = [], [] + for fk in self.foreign_keys: + if fk["ref_table"] != table: + continue + + # Count the number of rows for each key. + child_table = self.tables[fk["table"]].copy() + child_table["_dummy_"] = 0.0 + child_counts = child_table.groupby(fk["field"]).count().iloc[:, 0:1] + child_counts.columns = ["_tmp_"] + + # Merge the counts into the parent table + parent_table = self.tables[table] + parent_table = parent_table.set_index(fk["ref_field"]) + parent_table = parent_table.join(child_counts).reset_index() + + X.append(parent_table["_tmp_"].fillna(0.0).values) + columns.append({"source_col": {"table_name": fk['table'], "col_name": fk['field']}, + "row_map": fk, + "aggregation": "datatracer.how_lineage.entries_count" + }) + + return np.array(X).transpose(), columns + + def _get_aggregations(self, table): + """ + Get the foreign keys where the given table is the parent. + """ + X, columns = [], [] + for fk in self.foreign_keys: + if fk["ref_table"] != table: + continue + + for op, op_name, op_str in [ + (lambda x: x.sum(), "SUM", "datatracer.how_lineage.entries_sum"), + (lambda x: x.max(), "MAX", "datatracer.how_lineage.entries_max"), + (lambda x: x.min(), "MIN", "datatracer.how_lineage.entries_min"), + (lambda x: x.std(), "STD", "datatracer.how_lineage.entries_std"), + ]: + # Count the number of rows for each key. + child_table = self.tables[fk["table"]].copy() + if len(child_table.columns) <= 1: + continue + + child_counts = op(child_table.groupby(fk["field"])) + old_column_names = list(child_counts.columns) + child_counts.columns = ["%s(%s)" % (op_name, col_name) + for col_name in old_column_names] + + # Merge the counts into the parent table + parent_table = self.tables[table] + parent_table = parent_table.set_index(fk["ref_field"]) + parent_table = parent_table.join(child_counts).reset_index() + + for old_name, col_name in zip(old_column_names, child_counts.columns): + if parent_table[col_name].dtype.kind == "f": + X.append(parent_table[col_name].fillna(0.0).values) + columns.append({"source_col": {"table_name": fk['table'], + "col_name": old_name}, + "row_map": fk, + "aggregation": op_str + }) + + return np.array(X).transpose(), columns + + def backward(self, feature_importances): + """ + This function takes an array of `feature_importances` which corresponds + to the `X` matrix produced by the last call to `forward`. It returns a + mapping from fields to importance scores. + """ + return [(column, importance) + for column, importance in zip(self.columns, feature_importances)] diff --git a/datatracer/jsons/pipelines/datatracer.column_map.basic.json b/datatracer/jsons/pipelines/datatracer.column_map.basic.json index ad66877..9f328ec 100644 --- a/datatracer/jsons/pipelines/datatracer.column_map.basic.json +++ b/datatracer/jsons/pipelines/datatracer.column_map.basic.json @@ -1,5 +1,6 @@ { "primitives": [ + "datatracer.primary_key.BasicPrimaryKeySolver", "datatracer.foreign_key.StandardForeignKeySolver", "datatracer.column_map.BasicColumnMapSolver" ] diff --git a/datatracer/jsons/pipelines/datatracer.foreign_key.standard.json b/datatracer/jsons/pipelines/datatracer.foreign_key.standard.json index 0ed3ab6..266075b 100644 --- a/datatracer/jsons/pipelines/datatracer.foreign_key.standard.json +++ b/datatracer/jsons/pipelines/datatracer.foreign_key.standard.json @@ -1,5 +1,6 @@ { "primitives": [ + "datatracer.primary_key.BasicPrimaryKeySolver", "datatracer.foreign_key.StandardForeignKeySolver" ] } diff --git a/datatracer/jsons/pipelines/datatracer.how_lineage.basic.json b/datatracer/jsons/pipelines/datatracer.how_lineage.basic.json new file mode 100644 index 0000000..ab9c8f6 --- /dev/null +++ b/datatracer/jsons/pipelines/datatracer.how_lineage.basic.json @@ -0,0 +1,7 @@ +{ + "primitives": [ + "datatracer.primary_key.BasicPrimaryKeySolver", + "datatracer.foreign_key.StandardForeignKeySolver", + "datatracer.how_lineage.BasicHowLineageSolver" + ] +} diff --git a/datatracer/jsons/primitives/datatracer.foreign_key.StandardForeignKeySolver.json b/datatracer/jsons/primitives/datatracer.foreign_key.StandardForeignKeySolver.json index 51d9518..2429d4d 100644 --- a/datatracer/jsons/primitives/datatracer.foreign_key.StandardForeignKeySolver.json +++ b/datatracer/jsons/primitives/datatracer.foreign_key.StandardForeignKeySolver.json @@ -6,7 +6,7 @@ "method": "fit", "args": [ { - "name": "list_of_databases", + "name": "dict_of_databases", "type": "list" } ] @@ -56,14 +56,6 @@ } }, "tunable": { - "threshold": { - "type": "float", - "default": 0.9, - "range": [ - 0.0, - 1.0 - ] - }, "n_estimators": { "type": "int", "default": 10, diff --git a/datatracer/jsons/primitives/datatracer.how_lineage.BasicHowLineageSolver.json b/datatracer/jsons/primitives/datatracer.how_lineage.BasicHowLineageSolver.json new file mode 100644 index 0000000..959939c --- /dev/null +++ b/datatracer/jsons/primitives/datatracer.how_lineage.BasicHowLineageSolver.json @@ -0,0 +1,134 @@ +{ + "name": "datatracer.how_lineage.BasicHowLineageSolver", + "description": "Detect the how-lineage of a column.", + "primitive": "datatracer.how_lineage.BasicHowLineageSolver", + "produce": { + "method": "solve", + "args": [ + { + "name": "tables", + "type": "dict" + }, + { + "name": "foreign_keys", + "type": "dict" + }, + { + "name": "target_table", + "default": null, + "type": "str" + }, + { + "name": "target_field", + "default": null, + "type": "str" + } + ], + "output": [ + { + "name": "how_lineage", + "type": "dict" + } + ] + }, + "hyperparameters": { + "fixed": { + "n_jobs": { + "type": "int", + "default": null + }, + "verbose": { + "type": "int", + "default": 0, + "range": [ + 0, + 100 + ] + }, + "warm_start": { + "type": "bool", + "default": false + } + }, + "tunable": { + "n_estimators": { + "type": "int", + "default": 10, + "range": [ + 1, + 500 + ] + }, + "criterion": { + "type": "str", + "default": "mse", + "values": [ + "mse", + "mae" + ] + }, + "max_features": { + "type": "str", + "default": "auto", + "range": [ + null, + "auto", + "log2", + "sqrt" + ] + }, + "max_depth": { + "type": "int", + "default": null, + "range": [ + 1, + 30 + ] + }, + "min_samples_split": { + "type": "int", + "default": 2, + "range": [ + 2, + 1000 + ] + }, + "min_samples_leaf": { + "type": "int", + "default": 1, + "range": [ + 1, + 1000 + ] + }, + "min_weight_fraction_leaf": { + "type": "float", + "default": 0.0, + "range": [ + 0.0, + 100.0 + ] + }, + "max_leaf_nodes": { + "type": "int", + "default": null + }, + "min_impurity_decrease": { + "type": "float", + "default": 0.0, + "range": [ + 0.0, + 10.0 + ] + }, + "bootstrap": { + "type": "bool", + "default": true + }, + "oob_score": { + "type": "bool", + "default": false + } + } + } +} diff --git a/datatracer/jsons/primitives/datatracer.primary_key.BasicPrimaryKeySolver.json b/datatracer/jsons/primitives/datatracer.primary_key.BasicPrimaryKeySolver.json index ff81725..c387f9f 100644 --- a/datatracer/jsons/primitives/datatracer.primary_key.BasicPrimaryKeySolver.json +++ b/datatracer/jsons/primitives/datatracer.primary_key.BasicPrimaryKeySolver.json @@ -6,7 +6,7 @@ "method": "fit", "args": [ { - "name": "list_of_databases", + "name": "dict_of_databases", "type": "list" } ] diff --git a/datatracer/pretrained/datatracer.column_map.basic.dt b/datatracer/pretrained/datatracer.column_map.basic.dt index 9497cf1..0aca85f 100644 Binary files a/datatracer/pretrained/datatracer.column_map.basic.dt and b/datatracer/pretrained/datatracer.column_map.basic.dt differ diff --git a/datatracer/pretrained/datatracer.foreign_key.standard.dt b/datatracer/pretrained/datatracer.foreign_key.standard.dt index 6fc8bf3..1d31361 100644 Binary files a/datatracer/pretrained/datatracer.foreign_key.standard.dt and b/datatracer/pretrained/datatracer.foreign_key.standard.dt differ diff --git a/datatracer/pretrained/datatracer.how_lineage.basic.dt b/datatracer/pretrained/datatracer.how_lineage.basic.dt new file mode 100644 index 0000000..92c3bb2 Binary files /dev/null and b/datatracer/pretrained/datatracer.how_lineage.basic.dt differ diff --git a/datatracer/pretrained/datatracer.primary_key.basic.dt b/datatracer/pretrained/datatracer.primary_key.basic.dt index 085df0e..4a92dfe 100644 Binary files a/datatracer/pretrained/datatracer.primary_key.basic.dt and b/datatracer/pretrained/datatracer.primary_key.basic.dt differ diff --git a/datatracer/primary_key/base.py b/datatracer/primary_key/base.py index f033f8f..df316d2 100644 --- a/datatracer/primary_key/base.py +++ b/datatracer/primary_key/base.py @@ -3,15 +3,15 @@ class PrimaryKeySolver(): - def fit(self, list_of_databases): + def fit(self, dict_of_databases): """Fit this solver. Args: - list_of_databases (list): - List of tuples containing ``MetaData`` instnces and table dictinaries, - which contain table names as input and ``pandas.DataFrames`` as values. + dict_of_databases (dict): + Map from database names to tuples containing ``MetaData`` + instances and table dictionaries, which contain table names + as input and ``pandas.DataFrames`` as values. """ - pass def solve(self, tables): """Solve the primary key detection problem. diff --git a/datatracer/primary_key/basic.py b/datatracer/primary_key/basic.py index ea46aaf..7e717ea 100644 --- a/datatracer/primary_key/basic.py +++ b/datatracer/primary_key/basic.py @@ -2,23 +2,26 @@ import numpy as np from sklearn.ensemble import RandomForestClassifier +from tqdm import tqdm from datatracer.primary_key.base import PrimaryKeySolver class BasicPrimaryKeySolver(PrimaryKeySolver): - def __init__(self, *args, **kwargs): + def __init__(self, threshold=[i / 20 for i in range(20)], *args, **kwargs): self._model_args = args self._model_kwargs = kwargs + self._threshold = threshold def _feature_vector(self, table, column_name): column = table[column_name] return [ list(table.columns).index(column_name), - list(table.columns).index(column_name) / len(table.columns), + 0.0 if len(table.columns) == 0 else list( + table.columns).index(column_name) / len(table.columns), 1.0 if column.nunique() == len(column) else 0.0, - column.nunique() / len(column), + 0.0 if len(column) == 0 else column.nunique() / len(column), 1.0 if "key" in column.name else 0.0, 1.0 if "id" in column.name else 0.0, 1.0 if "_key" in column.name else 0.0, @@ -27,41 +30,68 @@ def _feature_vector(self, table, column_name): 1.0 if column.dtype == "object" else 0.0, ] - def fit(self, list_of_databases): + def fit(self, dict_of_databases): """Fit this solver. Args: - list_of_databases (list): - List of tuples containing ``MetaData`` instnces and table dictinaries, - which contain table names as input and ``pandas.DataFrames`` as values. + dict_of_databases (dict): + Map from database names to tuples containing ``MetaData`` + instances and table dictionaries, which contain table names + as input and ``pandas.DataFrames`` as values. """ X, y = [], [] - for metadata, tables in list_of_databases: + iterator = tqdm(dict_of_databases.items()) + for database_name, (metadata, tables) in iterator: + iterator.set_description("Extracting features from %s" % database_name) for table in metadata.get_tables(): if "primary_key" not in table: - continue - if not isinstance(table["primary_key"], str): - continue + pk = [] + elif not isinstance(table["primary_key"], str): + pk = table["primary_key"] + else: + pk = [table["primary_key"]] - primary_key = table["primary_key"] + table["primary_key"] for column in tables[table["name"]].columns: X.append(self._feature_vector(tables[table["name"]], column)) - y.append(1.0 if primary_key == column else 0.0) + y.append(1.0 if column in pk else 0.0) X, y = np.array(X), np.array(y) self.model = RandomForestClassifier(*self._model_args, **self._model_kwargs) self.model.fit(X, y) + if isinstance(self._threshold, list): + best_f1 = -float('inf') + best_threshold = None + pred_y = self.model.predict(X) + len_true = sum(y) + for threshold in self._threshold: + filtered_y = (pred_y >= threshold).astype(float) + intersect = sum(filtered_y * y) + len_pred = sum(filtered_y) + if intersect * len_true * len_pred == 0: + f1 = 0 + else: + precision = intersect / len_pred + recall = intersect / len_true + f1 = 2.0 * precision * recall / (precision + recall) + if f1 > best_f1: + best_f1 = f1 + best_threshold = threshold + self._threshold = best_threshold + + def _score_all_keys(self, table): + return [(column, self.model.predict([self._feature_vector(table, column)])) + for column in table.columns] + def _find_primary_key(self, table): - best_column, best_score = None, float("-inf") - for column in table.columns: - score = self.model.predict([self._feature_vector(table, column)]) - if score > best_score: - best_column = column - best_score = score - - return best_column + ret = [] + for column, score in self._score_all_keys(table): + if score >= self._threshold: + ret.append(column) + + return ret def solve(self, tables): """Solve the problem. diff --git a/setup.py b/setup.py index 27307b7..eead0f5 100644 --- a/setup.py +++ b/setup.py @@ -12,6 +12,7 @@ history = history_file.read() install_requires = [ + 'boto3>=1.13,<2', 'pandas>=0.23.4,<0.25', 'scikit-learn>=0.20.0,<0.21', 'numpy<1.17,>=1.15.2', @@ -20,7 +21,7 @@ 'falcon>=2.0.0,<3', 'hug>=2.6.1,<3', 'pyyaml>=5.3.1,<6', - 'tqdm>=4.46.1,<5', + 'tqdm>=4,<5', ] setup_requires = [ @@ -62,6 +63,10 @@ # Advanced testing 'coverage>=4.5.1,<6', 'tox>=2.9.1,<4', + + # benchmarking + 'dask>=2.15,<3', + 'distributed>=2.15,<3', ] setup( @@ -84,7 +89,8 @@ 'pipelines=datatracer:MLBLOCKS_PIPELINES' ], 'console_scripts': [ - 'datatracer=datatracer.__main__:main' + 'datatracer=datatracer.__main__:main', + 'datatracer-benchmark=benchmark.benchmark:main' ], }, extras_require={ @@ -100,7 +106,7 @@ keywords='datatracer data-tracer Data Tracer', name='datatracer', packages=find_packages(include=['datatracer', 'datatracer.*']), - python_requires='>=3.5,<3.8', + python_requires='>=3.5', setup_requires=setup_requires, test_suite='tests', tests_require=tests_require, diff --git a/tests/test_how_lineage.py b/tests/test_how_lineage.py new file mode 100644 index 0000000..6f36910 --- /dev/null +++ b/tests/test_how_lineage.py @@ -0,0 +1,9 @@ +from unittest import TestCase + +from datatracer.how_lineage import HowLineageSolver + + +class TestColumnMap(TestCase): + + def test_A(self): + HowLineageSolver() diff --git a/tutorials/DataTracer Quickstart.ipynb b/tutorials/DataTracer Quickstart.ipynb index a421095..aff0311 100644 --- a/tutorials/DataTracer Quickstart.ipynb +++ b/tutorials/DataTracer Quickstart.ipynb @@ -33,7 +33,9 @@ { "cell_type": "code", "execution_count": 1, - "metadata": {}, + "metadata": { + "tags": [] + }, "outputs": [ { "name": "stdout", @@ -83,7 +85,7 @@ { "data": { "text/plain": [ - "dict_keys(['mutagenesis', 'Chess', 'posts', 'classicmodels', 'university', 'Bupa', 'trains', 'SameGen', 'NBA', 'pubs'])" + "dict_keys(['posts', 'NBA', 'university', 'pubs', 'Chess', 'classicmodels', 'mutagenesis', 'Bupa', 'trains', 'SameGen'])" ] }, "execution_count": 3, @@ -396,15 +398,6 @@ "the dataset that we just explored, using the rest of the datasets as our training data." ] }, - { - "cell_type": "code", - "execution_count": 10, - "metadata": {}, - "outputs": [], - "source": [ - "training_datasets = list(datasets.values())" - ] - }, { "cell_type": "markdown", "metadata": {}, @@ -424,7 +417,7 @@ }, { "cell_type": "code", - "execution_count": 11, + "execution_count": 10, "metadata": {}, "outputs": [ { @@ -433,13 +426,14 @@ "['datatracer.column_map.basic',\n", " 'datatracer.foreign_key.basic',\n", " 'datatracer.foreign_key.standard',\n", + " 'datatracer.how_lineage.basic',\n", " 'datatracer.metadata.update_metadata_column_map',\n", " 'datatracer.metadata.update_metadata_foreign_keys',\n", " 'datatracer.metadata.update_metadata_primary_keys',\n", " 'datatracer.primary_key.basic']" ] }, - "execution_count": 11, + "execution_count": 10, "metadata": {}, "output_type": "execute_result" } @@ -471,7 +465,7 @@ }, { "cell_type": "code", - "execution_count": 12, + "execution_count": 11, "metadata": {}, "outputs": [], "source": [ @@ -490,11 +484,19 @@ }, { "cell_type": "code", - "execution_count": 13, + "execution_count": 12, "metadata": {}, - "outputs": [], + "outputs": [ + { + "name": "stderr", + "output_type": "stream", + "text": [ + "Extracting features from SameGen: 100%|██████████| 9/9 [00:00<00:00, 43.14it/s] \n" + ] + } + ], "source": [ - "dtr.fit(training_datasets)" + "dtr.fit(datasets)" ] }, { @@ -507,7 +509,7 @@ }, { "cell_type": "code", - "execution_count": 14, + "execution_count": 13, "metadata": {}, "outputs": [], "source": [ @@ -523,23 +525,23 @@ }, { "cell_type": "code", - "execution_count": 15, + "execution_count": 14, "metadata": {}, "outputs": [ { "data": { "text/plain": [ - "{'customers': 'customerNumber',\n", - " 'employees': 'employeeNumber',\n", - " 'offices': 'officeCode',\n", - " 'orderdetails': 'orderNumber',\n", - " 'orders': 'orderNumber',\n", - " 'payments': 'customerNumber',\n", - " 'productlines': 'productLine',\n", - " 'products': 'productCode'}" + "{'customers': ['customerNumber'],\n", + " 'employees': ['employeeNumber'],\n", + " 'offices': ['officeCode'],\n", + " 'orderdetails': ['orderNumber'],\n", + " 'orders': ['orderNumber'],\n", + " 'payments': ['customerNumber'],\n", + " 'productlines': ['productLine'],\n", + " 'products': ['productCode']}" ] }, - "execution_count": 15, + "execution_count": 14, "metadata": {}, "output_type": "execute_result" } @@ -561,20 +563,21 @@ }, { "cell_type": "code", - "execution_count": 16, + "execution_count": 15, "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ - "extracting features: 100%|██████████| 9/9 [00:01<00:00, 8.93it/s]\n" + "Extracting features from SameGen: 100%|██████████| 9/9 [00:00<00:00, 46.95it/s] \n", + "Extracting features from SameGen: 100%|██████████| 9/9 [00:00<00:00, 20.83it/s] \n" ] } ], "source": [ "dtr = DataTracer('datatracer.foreign_key.standard')\n", - "dtr.fit(training_datasets)\n", + "dtr.fit(datasets)\n", "foreign_keys = dtr.solve(tables)" ] }, @@ -588,13 +591,17 @@ }, { "cell_type": "code", - "execution_count": 17, + "execution_count": 16, "metadata": {}, "outputs": [ { "data": { "text/plain": [ - "[{'table': 'payments',\n", + "[{'table': 'products',\n", + " 'field': 'productLine',\n", + " 'ref_table': 'productlines',\n", + " 'ref_field': 'productLine'},\n", + " {'table': 'payments',\n", " 'field': 'customerNumber',\n", " 'ref_table': 'customers',\n", " 'ref_field': 'customerNumber'},\n", @@ -616,7 +623,7 @@ " 'ref_field': 'officeCode'}]" ] }, - "execution_count": 17, + "execution_count": 16, "metadata": {}, "output_type": "execute_result" } @@ -647,20 +654,21 @@ }, { "cell_type": "code", - "execution_count": 18, + "execution_count": 17, "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ - "extracting features: 100%|██████████| 9/9 [00:01<00:00, 9.00it/s]\n" + "Extracting features from SameGen: 100%|██████████| 9/9 [00:00<00:00, 43.29it/s] \n", + "Extracting features from SameGen: 100%|██████████| 9/9 [00:00<00:00, 21.37it/s] \n" ] } ], "source": [ "dtr = DataTracer('datatracer.column_map.basic')\n", - "dtr.fit(training_datasets)\n", + "dtr.fit(datasets)\n", "column_map = dtr.solve(\n", " tables,\n", " target_table='orderdetails',\n", @@ -678,20 +686,18 @@ }, { "cell_type": "code", - "execution_count": 19, - "metadata": { - "scrolled": true - }, + "execution_count": 18, + "metadata": {}, "outputs": [ { "data": { "text/plain": [ - "{('orderdetails', 'orderNumber'): 0.3810398229864858,\n", - " ('orderdetails', 'priceEach'): 0.4231615036348897,\n", - " ('orderdetails', 'orderLineNumber'): 0.19579867337862447}" + "{('orderdetails', 'orderNumber'): 0.3867085505946312,\n", + " ('orderdetails', 'priceEach'): 0.4226638356684435,\n", + " ('orderdetails', 'orderLineNumber'): 0.19062761373692533}" ] }, - "execution_count": 19, + "execution_count": 18, "metadata": {}, "output_type": "execute_result" } @@ -717,7 +723,7 @@ }, { "cell_type": "code", - "execution_count": 20, + "execution_count": 19, "metadata": {}, "outputs": [], "source": [ @@ -733,7 +739,7 @@ }, { "cell_type": "code", - "execution_count": 21, + "execution_count": 20, "metadata": {}, "outputs": [], "source": [ @@ -746,19 +752,16 @@ }, { "cell_type": "code", - "execution_count": 22, + "execution_count": 21, "metadata": {}, "outputs": [ { "data": { "text/plain": [ - "{('orderdetails', 'orderNumber'): 0.00020721216199988827,\n", - " ('orderdetails', 'priceEach'): 0.00010550603910546229,\n", - " ('orderdetails', 'orderLineNumber'): 4.512101354345987e-05,\n", - " ('orderdetails', 'quantityOrdered_x2'): 0.9996421607853512}" + "{('orderdetails', 'quantityOrdered_x2'): 0.9996667966308828}" ] }, - "execution_count": 22, + "execution_count": 21, "metadata": {}, "output_type": "execute_result" } @@ -806,7 +809,7 @@ }, { "cell_type": "code", - "execution_count": 23, + "execution_count": 22, "metadata": {}, "outputs": [], "source": [ @@ -830,7 +833,7 @@ }, { "cell_type": "code", - "execution_count": 24, + "execution_count": 23, "metadata": {}, "outputs": [ { @@ -844,7 +847,7 @@ " {'table': 'orderdetails', 'field': 'orderLineNumber'}]}]" ] }, - "execution_count": 24, + "execution_count": 23, "metadata": {}, "output_type": "execute_result" } @@ -870,7 +873,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.6.9" + "version": "3.7.4" } }, "nbformat": 4, diff --git a/tutorials/Introducing DataTracer.ipynb b/tutorials/Introducing DataTracer.ipynb index 09504d1..c2b5717 100644 --- a/tutorials/Introducing DataTracer.ipynb +++ b/tutorials/Introducing DataTracer.ipynb @@ -265,7 +265,7 @@ { "data": { "text/plain": [ - "{'users': 'id', 'posts': 'id'}" + "{'users': ['id'], 'posts': ['id', 'uid']}" ] }, "execution_count": 5, @@ -338,12 +338,7 @@ { "data": { "text/plain": [ - "{('users', 'id'): 0.0,\n", - " ('users', 'birthyear'): 0.999998434445148,\n", - " ('users', 'height'): 1.017586199833654e-06,\n", - " ('users', 'nb_posts'): 0.0,\n", - " ('posts', 'uid'): 0.0,\n", - " ('posts', 'id'): 0.0}" + "{('users', 'birthyear'): 0.9999985726528566}" ] }, "execution_count": 7, @@ -379,12 +374,7 @@ { "data": { "text/plain": [ - "{('users', 'id'): 0.0,\n", - " ('users', 'age'): 0.0,\n", - " ('users', 'birthyear'): 0.0,\n", - " ('users', 'height'): 0.0,\n", - " ('posts', 'uid'): 0.44796541862508577,\n", - " ('posts', 'id'): 0.5470063327357171}" + "{('posts', 'uid'): 1.0}" ] }, "execution_count": 8, @@ -421,7 +411,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.6.9" + "version": "3.7.4" } }, "nbformat": 4,