Greg/test data export main

platform/micro-services/command-line-interface/source/cli/peoples-speech.py

@@ -3,26 +3,42 @@
 
 from argparse import ArgumentParser
 
-import peoples_speech.data_book
-import peoples_speech.data_export
-import peoples_speech.task_manager
-
 import config
 
 import logging
 
+import peoples_speech.data_book
+import peoples_speech.data_export
+import peoples_speech.task_manager
+
 logger = logging.getLogger(__name__)
 
 def main():
     parser = ArgumentParser("The MLCommons data engineering framework.")
 
+    subparsers = parser.add_subparsers()
+
+    setup_cli_parser(subparsers)
+    peoples_speech.data_export.setup_cli_parser(subparsers)
+
+    args = parser.parse_args()
+
+    args.func(args)
+
+def setup_cli_parser(subparsers):
+
+    parser = subparsers.add_parser('dataset')
+
     parser.add_argument("-i", "--data-book-path", default=sample_databook_path(), help="Path to data book to generate a dataset for.")
     parser.add_argument("-o", "--output-dataset-path", default="", help="The path to save the new dataset.")
-    parser.add_argument("-c", "--config-file-path", default=".csv", help="The path to save the new dataset.")
+    parser.add_argument("-c", "--config-file-path", default=".csv", help="The path to the config file.")
     parser.add_argument("-v", "--verbose", default=False, action="store_true", help="Print out debug messages.")
     parser.add_argument("-vi", "--verbose-info", default=False, action="store_true", help="Print out info messages.")
 
-    arguments = vars(parser.parse_args())
+    parser.set_defaults(func=dispatch)
+
+def dispatch(args):
+    arguments = vars(args)
 
     config = setup_config(arguments)
 

platform/micro-services/data-export/build-scripts/flask/cloud-build/cloudbuild.sh

@@ -0,0 +1,13 @@
+#! /bin/bash
+
+# Safely execute this bash script
+# e exit on first failure
+# u unset variables are errors
+# f disable globbing on *
+# pipefail | produces a failure code if any stage fails
+set -euf -o pipefail
+
+# Get the directory of this script
+LOCAL_DIRECTORY="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+
+gcloud builds submit --config cloudbuild.yaml $LOCAL_DIRECTORY/../../../../../..

platform/micro-services/data-export/build-scripts/flask/cloud-build/cloudbuild.yaml

@@ -0,0 +1,14 @@
+steps:
+# build the data export container
+- name: 'gcr.io/cloud-builders/docker'
+  args: [ 'build', '-t', 'gcr.io/the-peoples-speech/data-export:0.12', '-f', 'platform/micro-services/data-export/build-scripts/flask/docker/Dockerfile', 'platform/micro-services/data-export' ]
+# push container image
+- name: "gcr.io/cloud-builders/docker"
+  args: ["push", "gcr.io/the-peoples-speech/data-export:0.12"]
+# deploy container image to GKE staging
+- name: "gcr.io/cloud-builders/gke-deploy"
+  args:
+  - run
+  - --filename=platform/micro-services/data-export/build-scripts/flask/kubernetes/export-service.yaml
+  - --location=europe-west4-c
+  - --cluster=peoples-speech-platform

platform/micro-services/data-export/build-scripts/flask/docker/Dockerfile

@@ -0,0 +1,9 @@
+FROM python
+
+COPY . /app
+
+ENV BOTO_CONFIG=/app/source/configs/google.boto
+
+EXPOSE 5000
+
+CMD /app/start-production $PORT

platform/micro-services/data-export/build-scripts/flask/docker/build-container.sh

@@ -0,0 +1,14 @@
+#! /bin/bash
+
+# Safely execute this bash script
+# e exit on first failure
+# u unset variables are errors
+# f disable globbing on *
+# pipefail | produces a failure code if any stage fails
+set -euf -o pipefail
+
+# Get the directory of this script
+LOCAL_DIRECTORY="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+
+docker build -t data-export:0.1 -f $LOCAL_DIRECTORY/Dockerfile $LOCAL_DIRECTORY/../../../../../..
+

platform/micro-services/data-export/build-scripts/flask/kubernetes/deploy-staging.sh

@@ -0,0 +1,2 @@
+gcloud container clusters get-credentials peoples-speech-platform
+kubectl create deployment data-export --image=gcr.io/peoples-speech/data-export:latest

platform/micro-services/data-export/build-scripts/flask/kubernetes/export-service.yaml

@@ -0,0 +1,50 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  labels:
+    app: data-export
+  name: data-export
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: data-export
+  template:
+    metadata:
+      labels:
+        app: data-export
+    spec:
+      containers:
+      - image: gcr.io/the-peoples-speech/data-export:0.12
+        name: data-export
+        ports:
+        - containerPort: 5000
+          name: tcp-c-5000
+        env:
+        - name: "PORT"
+          value: "5000"
+        volumeMounts:
+          - mountPath: "/app/credentials"
+            name: gcloud-service-account-key
+            readOnly: true
+      volumes:
+        - name: gcloud-service-account-key
+          secret:
+            secretName: gcloud-service-account-key
+---
+apiVersion: v1
+kind: Service
+metadata:
+  labels:
+    app: data-export
+  name: data-export
+spec:
+  selector:
+    app: data-export
+  ports:
+    - port: 5000
+      targetPort: 5000
+      protocol: TCP
+      name: tcp-5000
+  type: LoadBalancer
+

platform/micro-services/data-export/build-scripts/flask/kubernetes/run-export-service.yaml

@@ -0,0 +1,21 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  labels:
+    app.kubernetes.io/name: data-export
+  name: data-export
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: data-export
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/name: data-export
+    spec:
+      containers:
+      - image: gcr.io/peoples-speech/data-export
+        name: data-export
+        ports:
+        - containerPort: 8080

platform/micro-services/data-export/requirements.txt

@@ -0,0 +1,7 @@
+flask
+flask_cors
+python-configuration[yaml]
+smart_open[gcs]
+redis
+gsutil
+google-compute-engine

platform/micro-services/data-export/run-tests

@@ -0,0 +1,41 @@
+#! /bin/bash
+
+# Safely execute this bash script
+# e exit on first failure
+# u unset variables are errors
+# f disable globbing on *
+# pipefail | produces a failure code if any stage fails
+set -euf -o pipefail
+
+# Get the directory of this script
+LOCAL_DIRECTORY="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+
+# Setup virtual environment
+PYTHON_ENV=$(python3 -c "import sys; sys.stdout.write(sys.prefix) if (hasattr(sys, 'real_prefix') or sys.base_prefix != sys.prefix) else sys.stdout.write('0')")
+if [[ $PYTHON_ENV == 0 ]];
+then
+echo "Not in virtual environment"
+
+ACTIVATE=$LOCAL_DIRECTORY/environment/bin/activate
+
+if [ ! -f $ACTIVATE ]; then
+echo "Virtual environment doesn't exist, making it..."
+python3 -m venv $LOCAL_DIRECTORY/environment
+python3 -m pip install --upgrade pip > /dev/null
+fi
+
+source $ACTIVATE
+else
+echo "Running in virtual environment $PYTHON_ENV"
+fi
+
+# Make sure requirements are installed
+pip install -r $LOCAL_DIRECTORY/requirements.txt > /dev/null
+
+# Set python environment
+PYTHONPATH+="$LOCAL_DIRECTORY/source"
+export PYTHONPATH
+
+python source/data_export/api/test/test_export_dataset.py
+
+

platform/micro-services/data-export/source/configs/default.yaml

@@ -0,0 +1,6 @@
+exporter:
+    type: CloudExporter
+    endpoint: http://34.91.68.228
+
+verbose: True
+

platform/micro-services/data-export/source/configs/local.yaml

@@ -0,0 +1,10 @@
+
+datasets:
+    "0": gs://the-peoples-speech-west-europe/peoples-speech-v0.8/unittest.csv
+
+exporter:
+    type: LocalExporter
+
+output_dataset_path: gs://the-peoples-speech-west-europe/peoples-speech-v0.8/unittest.tar.gz
+
+verbose: True

platform/micro-services/data-export/source/configs/server.yaml

@@ -0,0 +1,10 @@
+exporter:
+    type: GoogleCloudParallelExporter
+    task_count: 1024
+
+google:
+    work_queue:
+        name: "export_work_queue"
+
+verbose: True
+

platform/micro-services/data-export/source/data_export/__init__.py

@@ -1,5 +1,7 @@
 
 from data_export.api.save_dataset import save_dataset
-
+from data_export.api.export_dataset_by_id import export_dataset_by_id
+from data_export.api.export_dataset import export_dataset
+from data_export.api.setup_cli_parser import setup_cli_parser
 
 

platform/micro-services/data-export/source/data_export/api/export_dataset.py

@@ -0,0 +1,13 @@
+
+from data_export.export import ExporterFactory
+from data_export.utility.get_config import get_config
+
+<<<<<<< HEAD
+def export_dataset(output_dataset, dataset, config = get_config()):
+    exporter = ExporterFactory(config).create()
+    return exporter.export(output_dataset, dataset)
+=======
+def export_dataset(dataset, config = get_config()):
+    exporter = ExporterFactory(config).create()
+    exporter.export(dataset)
+>>>>>>> d17bf137... TEST: Added unit test for data export.

platform/micro-services/data-export/source/data_export/api/export_dataset_by_id.py

@@ -0,0 +1,10 @@
+
+from data_export.export import ExporterFactory
+from data_export.database import get_dataset_from_id
+from data_export.utility.get_config import get_config
+
+def export_dataset_by_id(dataset_id, config = get_config()):
+    exporter = ExporterFactory(config).create()
+    dataset = get_dataset_from_id(config, dataset_id)
+    exporter.export(dataset)
+

platform/micro-services/data-export/source/data_export/api/setup_cli_parser.py

@@ -0,0 +1,70 @@
+from data_export.api.export_dataset import export_dataset
+
+import config
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+
+
+def setup_cli_parser(subparsers):
+
+    parser = subparsers.add_parser('export')
+
+    parser.add_argument("-i", "--input-dataset", default="gs://the-peoples-speech-west-europe/peoples-speech-v0.8/unittest.csv", help="The dataset to export.")
+    parser.add_argument("-o", "--output-dataset-path", default="gs://the-peoples-speech-west-europe/peoples-speech-v0.8/unittest.tar.gz", help="The path to save the new dataset.")
+    parser.add_argument("-c", "--config-file-path", default="", help="The path to the config file.")
+    parser.add_argument("-v", "--verbose", default=False, action="store_true", help="Print out debug messages.")
+    parser.add_argument("-vi", "--verbose-info", default=False, action="store_true", help="Print out info messages.")
+
+    parser.set_defaults(func=dispatch)
+
+def dispatch(args):
+    arguments = vars(args)
+
+    config = setup_config(arguments)
+
+    setup_logging(config)
+
+    logger.debug("Full config: " + str(config))
+
+    export_dataset(config["output_dataset_path"], config["input_dataset"], config)
+
+def setup_config(dictionary):
+    return config.ConfigurationSet(
+        config.config_from_env(prefix="MLCOMMONS"),
+        config.config_from_yaml(config_path(dictionary), read_from_file=True),
+        config.config_from_dict(dictionary),
+    )
+
+def config_path(dictionary):
+    if os.path.exists(dictionary["config_file_path"]):
+        return dictionary["config_file_path"]
+
+    home = os.path.expanduser("~")
+    home_config_path = os.path.join(home, ".mlcommons", "config.yaml")
+    if os.path.exists(home_config_path):
+        return home_config_path
+
+    return os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "configs", "default.yaml")
+
+def setup_logging(arguments):
+
+    logging_format = "%(asctime)s - %(levelname)s - %(name)s - %(message)s"
+
+    if arguments["verbose"]:
+        logging.basicConfig(level=logging.DEBUG, format=logging_format)
+    elif arguments["verbose_info"]:
+        logging.basicConfig(level=logging.INFO, format=logging_format)
+    else:
+        logging.basicConfig(level=logging.WARNING, format=logging_format)
+
+    root_logger = logging.getLogger()
+
+    if arguments["verbose"]:
+        root_logger.setLevel(logging.DEBUG)
+    elif arguments["verbose_info"]:
+        root_logger.setLevel(logging.INFO)
+    else:
+        root_logger.setLevel(logging.WARNING)
+

platform/micro-services/data-export/source/data_export/api/test/test_export_dataset.py

@@ -0,0 +1,36 @@
+import unittest
+import tarfile
+import io
+from data_export.api.export_dataset import export_dataset
+from data_export.utility.load_dataset_csv import load_dataset_csv
+from data_export.utility.load_dataset_csv import load_dataset_csv_from_file
+
+class TestDataExport(unittest.TestCase):
+
+    def test_tiny_local(self):
+        tiny_dataset = "gs://the-peoples-speech-west-europe/peoples-speech-v0.8/unittest.csv"
+        tiny_config = {
+            "exporter" : { "type" : "LocalExporter" } }
+
+        export_dataset("/tmp/export-data/unittest.tar.gz", tiny_dataset, tiny_config)
+
+        self.extract_dataset(tiny_config)
+        self.verify_dataset(tiny_dataset, tiny_config)
+
+    def extract_dataset(self, config):
+        with tarfile.open(config["output_dataset_path"], "r:gz") as tar:
+            members = tar.getmembers()
+
+    def verify_dataset(self, dataset, config):
+
+        samples = load_dataset_csv(dataset, task_id=0, task_count=1)
+        with tarfile.open(config["output_dataset_path"], "r:gz") as tar:
+            samples_file = io.TextIOWrapper(tar.extractfile("export-data/dataset.csv"))
+            loaded_samples = load_dataset_csv_from_file(samples_file, task_id=0, task_count=1)
+
+            for original_sample, new_sample in zip(samples, loaded_samples):
+                self.assertEqual(original_sample, new_sample)
+
+
+if __name__ == '__main__':
+    unittest.main()