ricardolsmendes / ci-cd-for-data-processing-workflow

This repository contains source code based on a guide on how to use Cloud Build and Cloud Composer to create a CI/CD pipeline for building, testing, and deploying a data processing workflow

Hey there, I am Ricardo Mendes.

With over 20 years of consulting and software engineering experience across multiple industries, I have developed a passion for data and its potential to drive innovation and transformation. I now lead data teams that leverage the latest technologies to boost the delivery of business value, chiefly performing customer-facing roles.

My technical background includes Python, Java, Google Cloud, AWS, and DevOps, as well as Data Governance, Metadata Management, Data Lineage, and emerging fields such as DataOps and MLOps.

---

You can also reach me at:

ci-cd-for-data-processing-workflow's Issues

GoogleCloudStorageDownloadOperator raises TypeError on Composer

The GoogleCloudStorageDownloadOperator (used by the CompareXComMapsOperator in wordcount_test.py) works with Python 2.7 but raises a TypeError: Object of type 'bytes' is not JSON serializable when running in Python 3.6.

The error is easily reproducible in the below Cloud Composer Environment:

1. Composer version: 1.16.8
2. Airflow version: 1.10.15
3. Python version: 3.6

Error details:

Traceback (most recent call last)
  File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 985, in _run_raw_tas
    result = task_copy.execute(context=context
  File "/usr/local/lib/airflow/airflow/contrib/operators/gcs_download_operator.py", line 96, in execut
    context['ti'].xcom_push(key=self.store_to_xcom_key, value=file_bytes
  File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 1516, in xcom_pus
    execution_date=execution_date or self.execution_date
  File "/usr/local/lib/airflow/airflow/utils/db.py", line 74, in wrappe
    return func(*args, **kwargs
  File "/usr/local/lib/airflow/airflow/models/xcom.py", line 101, in se
    value = XCom.serialize_value(value
  File "/usr/local/lib/airflow/airflow/models/xcom.py", line 224, in serialize_valu
    return json.dumps(value).encode('UTF-8'
  File "/opt/python3.6/lib/python3.6/json/__init__.py", line 231, in dump
    return _default_encoder.encode(obj
  File "/opt/python3.6/lib/python3.6/json/encoder.py", line 199, in encod
    chunks = self.iterencode(o, _one_shot=True
  File "/opt/python3.6/lib/python3.6/json/encoder.py", line 257, in iterencod
    return _iterencode(o, 0
  File "/opt/python3.6/lib/python3.6/json/encoder.py", line 180, in defaul
    o.__class__.__name__
TypeError: Object of type 'bytes' is not JSON serializabl

GoogleCloudStorageDownloadOperator is deprecated and probably needs to be replaced in order to fix the test workflow.