Fetching the Data

For our mailing list example, we use data from public archives on the internet. Ideally, you want to connect to a database, stream, or other data repository. However, even in production, fetching web data can be necessary. First, we’ll implement our data fetching algorithm, which takes an Apache Software Foundation (ASF) project’s email list location along with the year from which to fetch messages. Example 5-1 returns the path to the records it fetches so we can use that as the input to the next pipeline stage.

def download_data(year: int) -> str:

  # The imports are inline here so Kubeflow can serialize the function correctly.
  from datetime import datetime
  from lxml import etree
  from requests import get
  from time import sleep

 import json

  def scrapeMailArchives(mailingList: str, year: int, month: int):
      #Ugly xpath code goes here. See the example repo if you're curious.

   datesToScrape =  [(year, i) for i in range(1,2)]

   records = []
   for y,m in datesToScrape:
     print(m,"-",y)
     records += scrapeMailArchives("spark-dev", y, m)
   output_path = '/data_processing/data.json'
   with open(output_path, 'w') as f:
     json.dump(records, f)

   return output_path

The above code downloads all of the mailing list data for a given year and saves it to a known path. In this example, a persistent volume needs to be mounted there to allow this data to move between stages, when we make our pipeline.

You may have a data dump as part of the machine learning pipeline, or a different system or team may provide one. For data on GCS or a PV, you can use the built-in components google-cloud/storage/download or filesystem/get_subdirectory to load the data instead of writing a custom function.

Data Cleaning: Filtering Out the Junk

Now that we’ve loaded our data, it’s time to do some simple data cleaning. Local tools are more common, so we’ll focus on them first. While data cleaning often depends on domain expertise, there are standard tools to assist with common tasks. A first step can be validating input records by checking the schema. That is to say; we check to see if the fields are present and are the right type.

To check the schema in the mailing list example, we ensure a sender, subject, and body all exist. To convert this into an independent component, we’ll make our function take a parameter for the input path and return the file path to the cleaned records. The amount of code it takes to do this is relatively small, shown in Example 5-2.

def clean_data(input_path: str) -> str:
    import json
    import pandas as pd

    print("loading records...")
    with open(input_path, 'r') as f:
        records = json.load(f)
    print("records loaded")

    df = pd.DataFrame(records)
    # Drop records without a subject, body, or sender
    cleaned = df.dropna(subset=["subject", "body", "from"])

    output_path_hdf = '/data_processing/clean_data.hdf'
    cleaned.to_hdf(output_path_hdf, key="clean")

    return output_path_hdf

There are many other standard data quality techniques besides dropping missing fields. Two of the more popular ones are imputing missing data ⁶ and analyzing/removing outliers which may be the result of incorrect measurements. Regardless of which additional general techniques you decide to perform, you can simply add them to your data cleaning function.

Domain specific data cleaning tools can also be beneficial. In the mailing list example, one potential source of noise in our data could be spam messages. One way to solve this would be by using Spamassassin. We can add this package to our container as in Example 5-3. Adding system software, not managed by pip, on top of the notebook images is a bit more complicated because of permissions. Most containers run as root, making it simple to install new system packages. However, because of Jupyter, the notebook containers run as a less privileged user. Installing new packages like this requires switching to the root user and back, which is not common in other Dockerfiles.

ARG base
FROM $base
# Run as root for updates
USER root
# Install Spamassassin
RUN apt-get update && 
    apt-get install -yq spamassassin spamc && 
    apt-get clean && 
    rm -rf /var/lib/apt/lists/* && 
    rm -rf /var/cache/apt
# Switch back to the expected user
USER jovyan

Once you’ve created this Dockerfile, you’ll want to build it and push the resulting image somewhere that the Kubeflow cluster can access as in Example 2-8.

Pushing a new container is not enough to let Kubeflow know that we want to use it. When constructing a pipeline stage with func_to_container_op you then need to specify the base_image parameter to the func_to_container_op function call. We’ll show this when we bring the example together as a pipeline in Example 5-35.

Here we see the power of containers again. You can add the tools we need on top of the building blocks provided by Kubeflow rather than making everything from scratch.

Once the data is cleaned, it’s time to make sure you have enough of it, or if not explore augmenting your data.

Formatting the Data

The correct format depends on which tool you’re using to do the feature preparation. If you’re sticking with the same tool you used for data preparation, an output can be the same as input. Otherwise, you might find this a good place to change formats. For example when using Spark for data prep and TensorFlow for training we often implement conversion to TFRecords here.

Feature Preparation

How to do feature preparation depends on the problem. With the mailing list example we can write all kinds of text-processing functions and combine them into features as shown in Example 5-4.

    df['domains'] = df['links'].apply(extract_domains)
    df['isThreadStart'] = df['depth'] == '0'

    # Arguably, you could split building the dataset away from the actual witchcraft.
    from sklearn.feature_extraction.text import TfidfVectorizer

    bodyV = TfidfVectorizer()
    bodyFeatures = bodyV.fit_transform(df['body'])

    domainV = TfidfVectorizer()

    ## A couple of "None" domains really screwed the pooch on this one.Also, no lists just space seperated domains.
    def makeDomainsAList(d):
        return ' '.join([a for a in d if not a is None])

    domainFeatures = domainV.fit_transform(
        df['domains'].apply(makeDomainsAList))

    from scipy.sparse import csr_matrix, hstack

    data = hstack([
        csr_matrix(df[[
            'containsPythonStackTrace', 'containsJavaStackTrace',
            'containsExceptionInTaskBody', 'isThreadStart'
        ]].to_numpy()), bodyFeatures, domainFeatures
    ])

So far, the example code is structured to allow you to turn each function into a separate pipeline stage, however other options exist. We’ll examine how to use the entire notebook as a pipeline stage in “Putting It Together in a Pipeline”.

There are data preparation tools beyond Notebooks and Python of course. Notebooks are not always the best tool as they can have difficulty with version control. Python doesn’t always have the libraries (or performance) you need. So we’ll now look at how to use other available tools.

Custom Containers

Pipelines are not just limited to notebooks or even to specific languages.⁷ Depending on the project, you may have a regular Python project, custom tooling, Python2, or even FORTRAN code as an essential component.

For instances, in Chapter 9 we will use Scala to perform one step in our pipeline, and present a sidebar on how to get started with a RStats container.

Sometimes you won’t be able to find a container that so closely matches your needs as we did here. In these cases, you can take a generic base image and build on top of it, which we look at more in .

Beyond the need for custom containers, another reason you might choose to move beyond notebooks is to explore distributed tools.

TensorFlow Extended

The TensorFlow community has created an excellent set of integrated tools for everything from data validation to model serving. At present, TFX’s data tools are all built on top of Apache Beam, which has the most support for distributed processing on Google Cloud. If you want to use Kubeflow’s TFX components, you are limited to a single node; this may change in future versions.

Kubeflow includes many of the TFX components in its pipeline system. TFX also has it’s own concept of pipelines. These are separate from Kubeflow pipelines, and in some cases TFX can be an alternative to Kubeflow. Here we will focus on the data and feature preparation components, since those are the simplest to be used with the rest of the Kubeflow ecosystem.

pip3 install tfx tensorflow-data-validation

tfx_csv_gen = kfp.components.load_component_from_file(
    "pipelines-0.2.5/components/tfx/ExampleGen/CsvExampleGen/component.yaml")
tfx_statistic_gen = kfp.components.load_component_from_file(
    "pipelines-0.2.5/components/tfx/StatisticsGen/component.yaml")
tfx_schema_gen = kfp.components.load_component_from_file(
    "pipelines-0.2.5/components/tfx/SchemaGen/component.yaml")
tfx_example_validator = kfp.components.load_component_from_file(
    "pipelines-0.2.5/components/tfx/ExampleValidator/component.yaml")

    fetch = kfp.dsl.ContainerOp(name='download',
                                image='busybox',
                                command=['sh', '-c'],
                                arguments=[
                                    'sleep 1;'
                                    'mkdir -p /tmp/data;'
                                    'wget ' + data_url +
                                    ' -O /tmp/data/results.csv'
                                ],
                                file_outputs={'downloaded': '/tmp/data'})
    # This expects a directory of inputs not just a single file

    records_example = tfx_csv_gen(input_base=fetch.output)

    stats = tfx_statistic_gen(input_data=records_example.output)
    schema_op = tfx_schema_gen(stats.output)

import tensorflow_data_validation as tfdv

schema = tfdv.load_schema_text("schema_info_2")
tfdv.display_schema(schema)

    tfx_example_validator(stats=stats.outputs['output'],
                          schema=schema_op.outputs['output'])

import tensorflow as tf
import tensorflow_transform as tft
from tensorflow_transform.tf_metadata import schema_utils

def preprocessing_fn(inputs):

    outputs = {}
    # TFT business logic goes here
    outputs["body_stuff"] = tft.compute_and_apply_vocabulary(inputs["body"],
                                                             top_k=1000)
    return outputs

    transformed_output = tfx_transform(
        input_data=records_example.output,
        schema=schema_op.outputs['output'],
        module_file=module_file)  # Path to your TFT code on GCS/S3

Spark Operators in Kubeflow

Using Kubeflow’s native spark-operator, EMR, or Dataproc are best once you’ve moved beyond the experimental phase. The most portable operator is the native spark-operator, which does not depend on any specific cloud. To use any of the operators, you need to package the Spark program and store it on either a distributed file system (like GCS, S3, etc.) or put it inside a container.

If you’re working in Python or R, we recommend building a Spark container so you can install your dependencies. With Scala or Java code, this is less critical. If you put the application inside of a container, you can reference it with local:///. You can use the gcr.io/spark-operator/spark-py:v2.4.5 container as the base or build your own container following Spark on Kubernetes instructions, or see Chapter 9. The Example 5-19 installs both any requirements and copies the application. If you decide to update the application, you can still use the container, just configure the main resource with a distributed filesystem. We cover building custom containers more in Chapter 9.

# Use the spark operator image as base
FROM gcr.io/spark-operator/spark-py:v2.4.5
# Install Python requirements
COPY requirements.txt /
RUN pip3 install -r /requirements.txt
# Now you can reference local:///job/my_file.py
RUN mkdir -p /job
COPY *.py /job

ENTRYPOINT ["/opt/entrypoint.sh"]

Two cloud-specific options for running Spark are the Amazon EMR or Google Dataproc components in Kubeflow. However, they each take different parameters meaning that you will need to translate your pipeline.

The EMR components allow you to set up clusters, submit jobs, and clean up the clusters. The two cluster task components are aws/emr/create_cluster for the start and aws/emr/delete_cluster. The component for running a PySpark job is aws/emr/submit_pyspark_job. If you are not re-using an external cluster, it’s important to trigger the delete component regardless of if the submit_pyspark_job component succeeds.

While they have different parameters, the workflow for Dataproc clusters mirrors that of EMR. The components are similarly named, with gcp/dataproc/create_cluster/ and gcp/dataproc/delete_cluster/ for the lifecycle and gcp/dataproc/submit_pyspark_job/ for running our job.

Unlike the EMR and Dataproc components, the Spark operator does not have a component. For Kubernetes operators without components, you can use the dsl.ResourceOp to call them. Using the ResourceOp to launch a Spark job is illustrated in Example 5-20.

resource = {
    "apiVersion": "sparkoperator.k8s.io/v1beta2",
    "kind": "SparkApplication",
    "metadata": {
        "name": "boop",
        "namespace": "kubeflow"
    },
    "spec": {
        "type": "Python",
        "mode": "cluster",
        "image": "gcr.io/boos-demo-projects-are-rad/kf-steps/kubeflow/myspark",
        "imagePullPolicy": "Always",
        # See the Dockerfile OR use GCS/S3/...
        "mainApplicationFile": "local:///job/job.py",
        "sparkVersion": "2.4.5",
        "restartPolicy": {
            "type": "Never"
        },
        "driver": {
            "cores": 1,
            "coreLimit": "1200m",
            "memory": "512m",
            "labels": {
                "version": "2.4.5",
            },
            # also try spark-operatoroperator-sa
            "serviceAccount": "spark-operatoroperator-sa",
        },
        "executor": {
            "cores": 1,
            "instances": 2,
            "memory": "512m"
        },
        "labels": {
            "version": "2.4.5"
        },
    }
}


@dsl.pipeline(name="local Pipeline", description="No need to ask why.")
def local_pipeline():

    rop = dsl.ResourceOp(
        name="boop",
        k8s_resource=resource,
        action="create",
        success_condition="status.applicationState.state == COMPLETED")

Reading the Input Data

Spark supports a wide variety of data sources, including (but not limited to): Parquet, JSON, JDBC, ORC, JSON, Hive, CSV, ElasticSearch, MongoDB, Neo4j, Cassandra, Snowflake, Redis, Riak Time Series, etc. ¹¹ Loading data is very straightforward, and often all that is needed is specifying the format. For example, in our mailing list example, reading the Parquet formatted output of our data preparation stage is done as in Example 5-25.

initial_posts = session.read.format("parquet").load(fs_prefix +
                                                    "/initial_posts")
ids_in_reply = session.read.format("parquet").load(fs_prefix + "/ids_in_reply")

If the above was instead formatted as JSON, we would only have to change “parquet” to “JSON.” ¹²

Validating the schema

We often believe we know the fields and types of our data. Spark can quickly discover the schema when our data is in a self-describing format like Parquet. In other formats, like JSON, the schema isn’t known until Spark reads the records. Regardless of the data format, it is good practice to specify the schema and ensure the data matches it, as shown in Example 5-26. Errors during data load are easier to debug than errors during model deployment.

ids_schema = StructType([
    StructField("In-Reply-To", StringType(), nullable=True),
    StructField("message-id", StringType(), nullable=True)
])
ids_in_reply = session.read.format("parquet").schema(ids_schema).load(
    fs_prefix + "/ids_in_reply")

You can configure Spark to handle corrupted and non-conforming records by dropping them, keeping them, or stopping the process (i.e. failing the job). The default is permissive, which keeps the invalid records while setting the fields to null, allowing us to handle schema mismatch with the same techniques for missing fields.

Handling missing fields

In many situations, some of our data is missing. You can choose to drop records with missing fields, fall back to secondary fields, impute averages, or leave as is. Spark’s built-in tools for these tasks are inside DataFrameNaFunctions. The correct solution depends on both your data and the algorithm you end up using. The most common is to drop records and make sure that we have not filtered out too many records, illustrated using the mailing list data in Example 5-27.

initial_posts_count = initial_posts.count()
initial_posts_cleaned = initial_posts.na.drop(how='any',
                                              subset=['body', 'from'])
initial_posts_cleaned_count = initial_posts_cleaned.count()

Filtering out bad data

Detecting incorrect data can be challenging. However, without performing at least some data cleaning, the model may train on noise. Often determining bad data depends on the practitioners domain knowledge of the problem.

A common technique supported in Spark is outlier removal. However, naively applying this can remove valid records. Using your domain experience, you can write a custom validation function and remove any records which don’t match it using the filter function in Spark as illustrated with our mailing list example in Example 5-28.

def is_ok(post):
    # Your special business logic goes here
    return True


spark_mailing_list_data_cleaned = spark_mailing_list_data_with_date.filter(
    is_ok)

Saving the output

Once you have the data ready, it’s time to save the output. If you’re going to use Apache Spark to do feature preparation, you can skip this step for now.

If you want to go back to single machine tools, it’s often simplest to save to a persistent volume. To do this, bring the data back to the main program by calling toPandas() as shown in Example 5-30. Now you can save the data in whatever format the next tool expects.

initial_posts.toPandas()

If the data is large, or you otherwise want to use an object store, Spark can write to many different formats (just as it can load from many different formats). The correct format depends on the tool you intend to use for feature preparation. Writing to parquet is shown in Example 5-31.

initial_posts.write.format("parquet").mode('overwrite').save(fs_prefix +
                                                             "/initial_posts")
ids_in_reply.write.format("parquet").mode('overwrite').save(fs_prefix +
                                                            "/ids_in_reply")

Now you’ve seen a variety of different tools you can use to source and clean the data. We’ve looked at the flexibility of local tools, the scalability of distributed tools, and the integration from TensorFlow Extended. With the data in shape, let’s now make sure the right features are available and get them in a usable format for the machine learning model.

Distributed Feature Preparation Using Apache Spark

Apache Spark has a large number of built-in feature preparation tools, in pyspark.ml.feature, that you can use to generate features. You can use Spark in the same way as you did during data preparation. You may find using Spark’s own ML pipeline an easy way to put together multiple feature preparation stages.

For the Spark mailing list example, we have textual input data. To allow us to train a variety of models, converting this into word vectors is our preferred form of feature prep. Doing so involves first tokenizing the data with Spark’s Tokenizer. Once we have the tokens, we can train a word2vec model and produce our word vectors. Example 5-32 illustrates how to prepare features for the mailing list example using Spark.

tokenizer = Tokenizer(inputCol="body", outputCol="body_tokens")
body_hashing = HashingTF(inputCol="body_tokens",
                         outputCol="raw_body_features",
                         numFeatures=10000)
body_idf = IDF(inputCol="raw_body_features", outputCol="body_features")
body_word2Vec = Word2Vec(vectorSize=5,
                         minCount=0,
                         numPartitions=10,
                         inputCol="body_tokens",
                         outputCol="body_vecs")
assembler = VectorAssembler(inputCols=[
    "body_features", "body_vecs", "contains_python_stack_trace",
    "contains_java_stack_trace", "contains_exception_in_task"
],
                            outputCol="features")

With this final distributed feature preparation example, you’re ready to scale up to handle larger data sizes if they ever come your way. If you’re working with smaller data, you’ve seen how you can use the same simple techniques of containerization to allow you to continue to work with your favourite tools. Either way, you’re almost ready for the next stage in the machine learning pipeline.

Using an Entire Notebook as a Dataprep Pipeline Stage

If you don’t want to turn the individual parts of the data preparation notebook into a pipeline, you can use the entire notebook as one stage. You can use the same containers used by JupyterHub to run the notebook programmatically.

To do this, you’ll need to make a new Dockerfile, specify that it is based on top of another container using FROM and then add a COPY directive to package the notebook inside the new container. Since the census data example has a pre-existing notebook, that’s the approach we’ve taken in Example 5-36.

FROM gcr.io/kubeflow-images-public/tensorflow-1.6.0-notebook-cpu

COPY ./ /workdir /

If you require additional Python dependencies, you can use the RUN directive to install them. Putting the dependencies in the container can help speed up the pipeline, especially for complicated packages. For our mailing list example, the Dockerfile would look like Example 5-37.

RUN pip3 install --upgrade lxml pandas

Now we can use this container like any other in the pipeline with dsl.ContainerOp as we did with the recommender example in the pipeline chapter. Now you have two ways to use notebooks in Kubeflow, and we’ll cover options beyond notebooks next.