Imagine an organization, which would process customer information from multiple sources to generate insights (or customer 360-degree view<\/strong>), which would help their customers to take better business decisions and act on them, for example:<\/p>\n\n\n\n Data unification aims to create a single source of truth<\/strong> for an organization’s data, which can improve data quality, enable better decision-making, and reduce the time and effort needed to access and analyze data. <\/p>\n\n\n\n Data unification is definitely a Big Data task. There are a lot of challenges you will have to deal with in order to implement a unified data platform. For example:<\/p>\n\n\n\n These challenges are difficult to handle using usual programming techniques such as creating a Java service to query data from multiple data sources and unify them via application logic. That is why we should choose a unique platform like Apache Spark that supports integration with a large number of systems and has the power to process large volumes of data so you can unify a massive amount of business data. Let dive dig into Spark in the next chapter.<\/p>\n\n\n\n The major problem of Big data is about dealing with large computations<\/strong> and large volumes of data<\/strong>. One particularly challenging area is data processing<\/strong>. Single machines do not have enough power and resources to perform computations on huge amounts of information (or the user probably does not have the time to wait for the computation to finish). <\/p>\n\n\n\n A cluster<\/strong> (or group, of computers) pools the resources of many machines together, allowing us Apache Spark is a unified computing engine and a set of libraries for parallel data processing<\/strong> on Spark is a huge part of companies solving massive-scale data challenges, from Resilient Distributed Dataset (RDD) is the fundamental data structure of Spark (or<\/strong> the backbone of PySpark<\/strong>). They are immutable Distributed collections of objects of any type and can reside on multiple nodes<\/strong> which Spark uses to achieve Faster and more efficient MapReduce<\/a> operations.<\/p>\n\n\n\n Spark has several core abstractions: Datasets, DataFrames, SQL Tables, and Resilient Distributed Datasets (RDDs)<\/strong>. These different abstractions all represent distributed collections of data. The easiest and most efficient are DataFrames, which are available in all languages.<\/strong><\/p>\n\n\n\n A DataFrame is the most common Structured API and represents a table of data with rows and The image below illustrates the fundamental difference: a spreadsheet sits on one computer in one specific location, whereas a Spark DataFrame can span thousands of computers. The reason for putting the data on more than one computer should be intuitive: either the data is too large to fit on one machine<\/strong> or it would simply take too long to perform that computation on one machine.<\/p>\n\n\n\n Since we control your Spark Application through a driver process<\/strong> called the SparkSession. The SparkSession instance is the way Spark executes user-defined manipulations across the cluster. There is a one-to-one correspondence between a SparkSession and a Spark Application. In Python, the variable is available as Spark when you start the console as below – a first Spark example locally.<\/p>\n\n\n\n You just ran your first Spark code! We created a DataFrame with one column containing 1,000 rows with values from 0 to 999. This range of numbers represents a distributed collection. When run on a cluster, each part of this range of numbers exists on a different executor. This is a Spark DataFrame.<\/p><\/blockquote><\/figure>\n\n\n\n How your code is actually executed across a Spark cluster? <\/p>\n\n\n\n To execute code, we must write code. This code is then submitted to Spark either through the console We will create a sample project to demonstrate how to implement data unification using Spark. This project simulates a Customer Data Platform with three databases:<\/p>\n\n\n\n The requirement of this project is to implement a “unify service”<\/strong> which creates a single customer 360-degree view unified from the above nht_contact <\/strong>and nht_payment <\/strong>data sources. This service requires a “primary_key”<\/strong> parameter, a set of fields in nht_contact.contact<\/strong> collection to identify which customer attributes will be used to match the data, when merging the data the latest data will be prioritized. <\/p>\n\n\n\n Below is a data unification example with 3 contacts and their activities, and payment data. After unifying with primate_key = [email, first_name], two unified contacts will be created:<\/p>\n\n\n\n PySpark will be used to interact with Spark cluster via Spark MongoDB Connector so you need to understand the basic fundamentals of Python programming API. Due to the complexity of building up Spark locally, we will use Docker to build up the entire local environment, so you need to install Docker on your local machine and understand the basic concept of Docker containerization. <\/p>\n\n\n\n The high-level architecture of this system is as below.<\/p>\n\n\n\n We will use Docker Compose to build the local environment including:<\/p>\n\n\n\n The unify service also needs to be containerized to Docker Image in order to run the entire system in Docker. Below is the content of the Dockerfile. Note that Spark requires Java runtime environment so we need to install Java as well.<\/p>\n\n\n\n We can containerize unify service to Dockeri mage with the below command.<\/p>\n\n\n\n The docker-compose file is as below. We use the “depend_on”<\/strong> attribute to make sure the unify-service image runs after the other services are ready. We also create sample data to demonstrate the unification including:<\/p>\n\n\n\n The project structure is quite simple and easy to understand. app.py<\/strong> is a Python application that implements the whole logic of the unification process and will be dived deeper into in the next chapters.<\/p>\n\n\n\n We will create the run_unify <\/strong>method to run the unification with a parameter is primary_key<\/strong> which is a list of contact fields we want to match.<\/p>\n\n\n\n We will create Spark Session with the Spark Master built up via Docker above. This Spark Session will connect to nht_contact <\/strong>databases as the input database and nht_unfied_contact <\/strong>as the output database using Spark Mongo Connector. Finally, we will load all documents in the “contact” collection into the Spark data frame to kick off the unification process.<\/p>\n\n\n\n After loading contact data to Spark data frame, we can use the groupby <\/strong>function to group these contacts by primary_key (1). Then we will load contact_activities<\/strong> data to a data frame because this collection is in the same Spark input database (2). Payment transaction data will be loaded via Python MySQL connector (3). Finally, we will perform the merge logic to ensure the latest data will be prioritized as well as combine contact_activies data, and payment data into the newly created unified contact (4).<\/p>\n\n\n\n The newly created unified contacts need to be converted to a data frame in order to utilize Spark’s powerful. We use the repartition method to distribute the workload of storing contact data across Spark workers which is a very important step.<\/p>\n\n\n\n We\u2019ve walked through the fundamental concepts of data unification as well as how data unification delivers value to the business side by creating a unified view of data from multiple sources. We’ve also looked at Spark’s introduction including Spark’s basic components and APIs to understand how Spark’s power can enable us to implement the data unification process.<\/p>\n\n\n\n Most noteworthy, we’ve implemented a comprehensive project to implement a simple unify_service which can be run in a container environment using Docker. It is always not an easy task due to the complexity of setting up Spark, especially building up Spark in the CI\/CD pipeline which is a very interesting topic to dive deeper into. Another state-of-the-art approach is using Spark in the cloud computing platform to reduce the development effort such as AWS EMR<\/a>. The demo project can be found on GitHub<\/a>.<\/p>\n","protected":false},"excerpt":{"rendered":" 1. The need of data unification Data unification refers to the process of combining and […]<\/p>\n","protected":false},"author":1,"featured_media":504,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[12],"tags":[46,45,30,43,44],"class_list":["post-345","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-big-data","tag-big-data","tag-data-unification","tag-docker","tag-python","tag-spark"],"yoast_head":"\n\n
![\"\"](\"https:\/\/thoaicoder.com\/wp-content\/uploads\/2023\/06\/image-4-1024x719.png\")
<\/figure>\n<\/div>\n\n\n\n
2. Spark introduction<\/h1>\n\n\n\n
2.1. What is Apache Spark?<\/h2>\n\n\n\n
to use all the cumulative resources as if they were a single computer. Now, a group of machines alone
is not powerful, you need a framework to coordinate work across them. Spark does just that,
managing and coordinating the execution of tasks on data across a cluster of computers (or nodes).<\/strong><\/p>\n\n\n![\"\"](\"https:\/\/thoaicoder.com\/wp-content\/uploads\/2023\/06\/image-3.png\")
<\/figure>\n<\/div>\n\n\n
computer clusters. Spark supports multiple widely used programming languages (Python, Java, Scala, and R), and includes libraries for diverse tasks ranging from SQL to streaming and machine learning.<\/p>\n\n\n\n
technology companies like Uber and Netflix using Spark\u2019s streaming and machine-learning tools, to
institutions like NASA, CERN, and the Broad Institute of MIT and Harvard applying Spark to
scientific data analysis<\/p>\n\n\n\n2.2. Spark concepts<\/h2>\n\n\n\n
2.2.1. RDD<\/h3>\n\n\n\n
2.2.2. DataFrame<\/h3>\n\n\n\n
columns<\/strong>. The list that defines the columns and the types within those columns is called the schema.
You can think of a DataFrame as a spreadsheet with named columns.<\/strong><\/p>\n\n\n\n![\"\"](\"https:\/\/thoaicoder.com\/wp-content\/uploads\/2023\/06\/image-5.png\")
<\/figure>\n\n\n\n2.2.3. Spark Session<\/h3>\n\n\n\n
from pyspark import SparkConf\nfrom pyspark.sql import SparkSession\n\nconf = SparkConf().setMaster(\"spark:\/\/localhost:7077\").setAppName(\"Hello\").set(\n \"spark.dynamicAllocation.enabled\", False).set(\n \"spark.driver.maxResultSize\", 0)\n\nmy_spark = SparkSession \\\n .builder \\\n .config(conf=conf) \\\n .getOrCreate()\n\nrange = my_spark.range(1000).toDF(\"number\")<\/code><\/pre>\n\n\n\n2.2.4. Overview of Structured API Execution<\/h3>\n\n\n\n
or via a submitted job. This code then passes through the Catalyst Optimizer<\/strong>, which decides how the
code should be executed (physical plan) and lays out a plan for doing so before, finally, the code is run and the result is returned to the user.<\/p>\n\n\n\n![\"\"](\"https:\/\/thoaicoder.com\/wp-content\/uploads\/2023\/06\/image-7.png\")
<\/figure>\n\n\n\n3. Data unification using PySpark<\/h1>\n\n\n\n
3.1. Introduction<\/h2>\n\n\n\n
\n
\n
\n
![\"\"](\"https:\/\/thoaicoder.com\/wp-content\/uploads\/2023\/07\/image-3-1024x539.png\")
<\/figure>\n\n\n\n![\"\"](\"https:\/\/thoaicoder.com\/wp-content\/uploads\/2023\/07\/image-1024x616.png\")
<\/figure>\n\n\n\n3.2. Build up Spark locally with Docker<\/h2>\n\n\n\n
\n
FROM python:3.8.2-buster\n\n# Install OpenJDK-11\nRUN apt-get update && \\\n apt-get install -y openjdk-11-jre-headless && \\\n apt-get clean;\n\n# Install PYTHON requirements\nWORKDIR \/srv\nADD .\/requirements.txt \/srv\/requirements.txt\nRUN pip install -r requirements.txt\nADD . \/srv\nCMD python \/srv\/app.py<\/code><\/pre>\n\n\n\ndocker build -t unify-service .<\/code><\/pre>\n\n\n\n\n
version: '3'\nservices:\n spark-master:\n image: bde2020\/spark-master:3.3.0-hadoop3.3\n container_name: spark-master\n ports:\n - \"8083:8080\"\n - \"7077:7077\"\n environment:\n - INIT_DAEMON_STEP=setup_spark\n spark-worker-1:\n image: bde2020\/spark-worker:3.3.0-hadoop3.3\n container_name: spark-worker-1\n depends_on:\n - spark-master\n ports:\n - \"8084:8081\"\n environment:\n - \"SPARK_MASTER=spark:\/\/spark-master:7077\"\n spark-worker-2:\n image: bde2020\/spark-worker:3.3.0-hadoop3.3\n container_name: spark-worker-2\n depends_on:\n - spark-master\n ports:\n - \"8085:8081\"\n environment:\n - \"SPARK_MASTER=spark:\/\/spark-master:7077\"\n contact-mongo-db:\n ports:\n - '27017:27017'\n container_name: contact-mongo-db\n restart: always\n logging:\n options:\n max-size: 1g\n environment:\n MONGO_INITDB_ROOT_USERNAME: nht\n MONGO_INITDB_ROOT_PASSWORD: abc123\n MONGO_INITDB_DATABASE: nht_contact\n image: mongo:6.0-focal\n volumes:\n - .\/init_mongo_contact\/mongo-init.js:\/docker-entrypoint-initdb.d\/mongo-init.js:ro\n healthcheck:\n test: \"exit 0\"\n unify-mongo-db:\n ports:\n - '27018:27017'\n container_name: unify-mongo-db\n restart: always\n logging:\n options:\n max-size: 1g\n environment:\n MONGO_INITDB_ROOT_USERNAME: nht\n MONGO_INITDB_ROOT_PASSWORD: abc123\n MONGO_INITDB_DATABASE: nht_unified_contact\n image: mongo:6.0-focal\n volumes:\n - .\/init_mongo_unify\/mongo-init.js:\/docker-entrypoint-initdb.d\/mongo-init.js:ro\n healthcheck:\n test: \"exit 0\"\n payment-mysql-db:\n container_name: payment-mysql-db\n platform: linux\/x86_64\n image: mysql:5.7\n restart: always\n environment:\n MYSQL_USER: 'nht'\n MYSQL_PASSWORD: 'abc123'\n MYSQL_ROOT_PASSWORD: 'abc123'\n ports:\n - '3316:3306'\n expose:\n - '3306'\n volumes:\n - .\/init_mysql_payment\/:\/docker-entrypoint-initdb.d\n healthcheck:\n test: \"exit 0\"\n unify-service:\n image: unify\n depends_on:\n spark-worker-1:\n condition: service_started\n spark-worker-2:\n condition: service_started\n contact-mongo-db:\n condition: service_healthy\n unify-mongo-db:\n condition: service_healthy\n payment-mysql-db:\n condition: service_healthy<\/code><\/pre>\n\n\n\n3.3. Project structure<\/h2>\n\n\n\n
![\"\"](\"https:\/\/thoaicoder.com\/wp-content\/uploads\/2023\/07\/image-1.png\")
<\/figure>\n\n\n\n3.4. Load contact data to Spark data frame<\/h2>\n\n\n\n
def run_unify(primary_key):\n conf = SparkConf().setMaster(\"spark:\/\/spark-master:7077\").setAppName(\"Unify\") \\\n .set(\"spark.dynamicAllocation.enabled\", False).set(\"spark.driver.maxResultSize\", \"3g\").set(\n \"spark.driver.memoryOverhead\", \"3g\").set(\"spark.executor.memory\", \"3g\")\n # initiate Spark session with input is contact database and output is unified contact database\n spark = SparkSession.builder \\\n .config(\"spark.jars.packages\", \"org.mongodb.spark:mongo-spark-connector_2.12:3.0.1\") \\\n .config(\"spark.mongodb.input.uri\",\n \"mongodb:\/\/nht:abc123@contact-mongo-db:27017\/nht_contact.contact?connectTimeoutMS=10000&authSource=admin&authMechanism=SCRAM-SHA-256\") \\\n .config(\"spark.mongodb.output.uri\",\n \"mongodb:\/\/nht:abc123@unify-mongo-db:27017\/nht_unified_contact.unified_contact?connectTimeoutMS=10000&authSource=admin&authMechanism=SCRAM-SHA-256\") \\\n .config(conf=conf) \\\n .getOrCreate()\n\n # load contact data to data frame\n df = spark.read.format(\"com.mongodb.spark.sql.DefaultSource\").load(schema=contact_schema)\n...<\/code><\/pre>\n\n\n\n3.5. Unify contact data<\/h2>\n\n\n\n
# 1. group contact data frame by primary key\ncolumns = [func.collect_list(x).alias(x) for x in df.columns]\nraw_unified_df = df.groupby(primary_key).agg(*columns)\nunified_contacts = []\n\n# 2. load contact activities to data frame\ncontact_activities = spark.read.format(\"com.mongodb.spark.sql.DefaultSource\") \\\n .option(\"collection\", \"contact_activity\") \\\n .load()\n\n# 3. load transactions from payment MySQL database\ntransactions = get_transactions();\n\n# 4. merge data to create unified contact\nfor row in raw_unified_df.rdd.collect():\n unified_contact = {}\n for attr, columns in row.__dict__.items():\n for column in columns:\n if column == \"_id\":\n column_values = []\n for items in row[column]:\n if items[0] is not None:\n column_values.append(items[0])\n # store id of contacts used to create this unified contact\n unified_contact[\"contact_ids\"] = column_values\n else:\n # merge data using latest value\n unified_contact[column] = row[column][-1]\n\n enrich_contact_activities(contact_activities, unified_contact)\n enrich_contact_payment(transactions, unified_contact)\n unified_contacts.append(unified_contact)<\/code><\/pre>\n\n\n\n3.6. Store unified contacts<\/h2>\n\n\n\n
# store unified contact to unified_contact database\nunified_contacts_df = spark.createDataFrame(unified_contacts, unified_contact_schema)\nunified_contacts_df = unified_contacts_df.repartition(100)\nunified_contacts_df.write.format(\"mongo\").mode(\"append\").save()\nspark.stop()<\/code><\/pre>\n\n\n\n4. Conclusion<\/h1>\n\n\n\n