Category: Our Insights

In today’s fast-evolving world, technology has redefined how we interact with our environments. From smart homes to automated workplaces, the demand for integrated systems that offer safety, convenience and enhanced user experience is higher than ever. As part of our ongoing mission to designs and build next-gen products, we recently took on a fascinating project: developing a mobile app (Android and iOS) for audio and video calls that works seamlessly with voice entry panels supporting VoIP (Voice over Internet Protocol).

The Client’s Vision: Safety, Accessibility and Innovation

The client approached us and requested an app that could serve as a comprehensive communication hub between residents, visitors, and building security systems. They wanted a mobile application that would:

• Enable smooth audio and video communication between users and VoIP-enabled voice entry panels.
• Improve building security by allowing residents to verify visitors before granting access.
• Increase convenience by integrating this communication into a single, easy-to-use app.
• Allow the mobile device to provide keyless entry.
• Should be available for both iOS and Android devices.
• In essence, the app would become the digital key to a smarter, safer living space.

Why VoIP? The Backbone of the Future

Voice over Internet Protocol (VoIP) has revolutionized modern communication by allowing voice and multimedia sessions over the internet. Its use in smart buildings, particularly in voice entry panels, provides several key benefits:

• Real-time communication: VoIP ensures fast, reliable, real-time communication between users and security systems.
• Cost-effectiveness: Unlike traditional landlines, VoIP uses existing internet infrastructure, lowering costs and providing more flexibility for building owners.
• Scalability: VoIP systems can easily be scaled up to accommodate new users and features as a building’s needs evolve.

By leveraging VoIP, we were able to design an app that offers superior connectivity, reliability and security.

Key Features of Our Audio/Video Call App

Our team focused on creating a solution that would offer a seamless user experience while enhancing security and control over building access. Below are the key features of the app we developed:

• Two-Way Audio/Video Communication

The app allows residents to receive calls from visitors at building entry panels, enabling real-time audio and video communication. This allows users to verify visitors’ identities visually and audibly, adding an extra layer of security.

• Remote Door Access

After verifying a visitor, the user can grant or deny access to the building via the app. This feature is particularly useful for those who may not be home but still need to authorize entry, such as in the case of package deliveries or housekeepers.

• VoIP-Based Calling

Since the app is built around VoIP technology, users benefit from high-quality calls that bypass traditional phone networks. The app integrates with voice entry panels that support VoIP, ensuring clear communication even over long distances.

• BLE Functionality: Turning Your Mobile Device into a Key

One of this app’s most groundbreaking features is the integration of Bluetooth Low Energy (BLE) technology, which allows the mobile device to function as a digital key. BLE is known for its ability to facilitate short-range communication with minimal power consumption, making it an ideal solution for mobile access control.

Instead of using traditional keycards, fobs, or physical keys, employees and residents can use their smartphones to open gates and doors. BLE communicates with the gate system within proximity, enabling automatic or one-tap entry. This improves convenience and reduces the need for physical keys, which can be lost or stolen.

• Cloud-Based Validation: Security in the Digital Age

To ensure that only authorized individuals can gain access, the app integrates with highly secure cloud servers for communication and validation. Every time a user attempts to open a gate or door using the app, the system verifies their credentials in real-time through cloud communication. This means that building managers or security teams can update or revoke access permissions instantly, without needing to reissue physical keys or cards.

Cloud-based validation offers multiple security layers, including encryption and authentication, ensuring that access control is both flexible and safe. It also allows for seamless monitoring and reporting, as every access attempt is logged in the system, providing valuable data for building management and security audits.

This feature also lets users manage communication logs, call history, and security footage from anywhere in the world. This feature is critical for building managers who want a real-time overview of all entry points.

• Streamlined Access for Employees and Residents

This app is designed with both employees and residents in mind. In office settings, employees can easily move through secured areas, using the app to access parking garages, secure floors, and meeting rooms. For residents of gated communities or apartment buildings, the app provides a hassle-free solution to enter gates and shared spaces like gyms or pools, without needing to fumble with keys or keycards and verifying visitors’ identities visually and audibly.

• Customizable Alerts and Notifications

Users can set up custom notifications for different types of events, such as when a delivery arrives or a family member enters the building. These alerts ensure that users are always aware of what’s happening around their living space.

• Scalability

This app can be scaled for use in various environments, from small residential buildings to large corporate complexes.

Enhancing the Future of Smart Buildings

This app’s development is part of a broader trend towards creating intelligent living environments where technology and security work hand-in-hand. Smart buildings equipped with VoIP-enabled voice entry panels, paired with mobile applications like ours not only enhance both security and user experience but also offer a glimpse into the future-a future where buildings are not just physical spaces, but adaptive, interactive ecosystems designed around the needs of their occupants.

We’re proud to contribute to this vision of safer, smarter, and more livable buildings. This project is a step forward in reshaping how residents interact with their living spaces, making life more convenient while ensuring peace of mind.

The Road Ahead

As we continue to refine and improve our app, we are excited about the potential to incorporate even more innovative features, such as Early Media Access, AI-driven facial recognition, advanced analytics for building managers and further integrations with home automation systems like Emergency Lighting System, Apartment Intercom System, Fire Detection and Alarm System including Smoke detectors, heat detectors and carbon monoxide detectors.

Our commitment remains strong: to create digital solutions which empower our clients and make the world a better, safer place to live in.

Stay tuned for more updates as we continue to push the boundaries of what smart building technology can achieve!

‍

Once upon a time, in the vast and ever-expanding world of data storage and processing, a new hero emerged. Its name? Delta Lake. This unsung champion was about to revolutionize the way organizations handled their data, and its journey was nothing short of remarkable.

The Need for a Data Savior

In this world, data was king, and it resided in various formats within the mystical realm of data lakes. Two popular formats, Parquet and Hive, had served their purposes well, but they harbored limitations that often left data warriors frustrated.

Enterprises faced a conundrum: they needed to make changes, updates, or even deletions to individual records within these data lakes. But it wasn’t as simple as it sounded. Modifying schemas was a perilous endeavor that could potentially disrupt the entire data kingdom.

Why? Because these traditional table formats lacked a vital attribute: ACID transactions. Without these safeguards, every change was a leap of faith.

The Rise of Delta Lake

Amidst this data turmoil, a new contender emerged: Delta Lake. It was more than just a format; it was a game-changer.

Delta Lake brought with it the power of ACID transactions. Every data operation within the kingdom was now imbued with atomicity, consistency, isolation, and durability. It was as if Delta Lake had handed data warriors an enchanted sword, making them invincible in the face of chaos.

But that was just the beginning of Delta Lake’s enchantment.

The Secrets of Delta Lake

Delta Lake was no ordinary table format; it was a storage layer that transcended the limits of imagination. It integrated seamlessly with Spark APIs, offering features that left data sorcerers in awe.

• Time Travel: Delta Lake allowed users to peer into the past, accessing previous versions of data. The transaction log became a portal to different eras of data history.
• Schema Evolution: It had the power to validate and evolve schemas as data changed. A shapeshifter of sorts, it embraced change effortlessly.
• Change Data Feed: With this feature, it tracked data changes at the granular level. Data sorcerers could now decipher the intricate dance of inserts, updates, and deletions.
• Data Skipping with Z-ordering: Delta Lake mastered the art of optimizing data retrieval. It skipped irrelevant files, ensuring that data requests were as swift as a summer breeze.
• DML Operations: It wielded the power of SQL-like data manipulation language (DML) operations. Updates, deletes, and merges were but a wave of its hand.

Delta Lake’s Allies

Delta Lake didn’t stand alone; it forged alliances with various data processing tools and platforms. Apache Spark, Apache Flink, Presto, Trino, Hive, DBT, and many others joined its cause. They formed a coalition to champion the cause of efficient data processing.

In the vast landscape of data management, Delta Lake stands as a beacon of innovation, offering a plethora of features that elevate your data handling capabilities to new heights. In this exhilarating adventure, we’ll explore the key features of Delta Lake and how they triumph over the limitations of traditional file formats, all while embracing the ACID properties.

ACID Properties: A Solid Foundation

In the realm of data, ACID isn’t just a chemical term; it’s a set of properties that ensure the reliability and integrity of your data operations. Let’s break down how Delta Lake excels in this regard.

A for Atomicity: All or Nothing

Imagine a tightrope walker teetering in the middle of their performance—either they make it to the other side, or they don’t. Atomicity operates on the same principle: either all changes happen, or none at all. In the world of Spark, this principle often takes a tumble. When a write operation fails midway, the old data is removed, and the new data is lost in the abyss. Delta Lake, however, comes to the rescue. It creates a transaction log, recording all changes made along with their versions. In case of a failure, data loss is averted, and your system remains consistent.

C for Consistency: The Guardians of Validity

Consistency is the gatekeeper of data validity. It ensures that your data remains rock-solid and valid at all times. Spark sometimes falters here. Picture this: your Spark job fails, leaving your system with invalid data remnants. Consistency crumbles. Delta Lake, on the other hand, is your data’s staunch guardian. With its transaction log, it guarantees that even in the face of job failure, data integrity is preserved.

I for Isolation: Transactions in Solitude

Isolation is akin to individual bubbles, where multiple transactions occur in isolation, without interfering with one another. Spark might struggle with this concept. If two Spark jobs manipulate the same dataset concurrently, chaos can ensue. One job overwrites the dataset while the other is still using it—no isolation, no guarantees. Delta Lake, however, introduces order into the chaos. Through its versioning system and transaction log, it ensures that transactions proceed in isolation, mitigating conflicts and ensuring the data’s integrity.

D for Durability: Unyielding in the Face of Failure

Durability means that once changes are made, they are etched in stone, impervious to system failures. Spark’s Achilles’ heel lies in its vulnerability to data loss during job failures. Delta Lake, however, boasts a different tale. It secures your data with unwavering determination. Every change is logged, and even in the event of job failure, data remains intact—a testament to true durability.

Time Travel: Rewriting the Past

Now, let’s embark on a fascinating journey through time. Delta Lake introduces a feature that can only be described as “time travel.” With this feature, you can revisit previous versions of your data, just like rewinding a movie. All of this magical history is stored in the transaction log, encapsulated within the mystical “_delta_log” folder. When you write data to a Delta table, it’s not just the present that’s captured; the past versions are meticulously preserved, waiting for your beck and call.

In conclusion, Delta Lake emerges as the hero of the data world, rewriting the rules of traditional file formats and conquering the challenges of the ACID properties. With its robust transaction log, versioning system, and the ability to traverse time, Delta Lake opens up a new dimension in data management. So, if you’re on a quest for data reliability, integrity, and a touch of magic, Delta Lake is your trusted guide through this thrilling journey beyond convention.

More Features of Delta Lake Are:

• UPSERT
• Schema Evolution
• Change Data Feed
• Data Skipping with Z-ordering
• DML Operations

The Quest for Delta Lake

Setting up Delta Lake was like embarking on a quest. Data adventurers ventured into the cloud, AWS, GCP, Azure, or even their local domains. They armed themselves with the delta-spark spells and summoned the JARs of delta-core, delta-contrib, and delta-storage, tailored to their Spark versions.

Requirements:

• Python
• Delta-spark
• Delta jars

You can configure in a Spark session and define the package name so it will be downloaded at the run time. As I said, I am using Spark version 3.3. We will require these things: delta-core, delta-contribs, delta-storage. You can download them from here: https://github.com/delta-io/delta/releases/ 

To use and configure various cloud storage options, there are separate .jars you can use: https://docs.delta.io/latest/delta-storage.html. Here, you can find .jars for AWS, GSC, and Azure to configure and use their data storage medium.

Run this command to install delta-spark first:

pip  install delta-spark

(If you are using dataproc or EMR, you can install this while creating cluster as a startup action, and if you are using serverless env like Glue or dataproc batches, you can create docker build or pass the .whl file for this package.)

You must also do this while downloading the .jar. If it is serverless, download the .jar, store it in cloud storage, like S3 or GS, and use that path while running the job. If it is a cluster like dataproc or EMR, you can download this on the cluster. 

One can also download these .jars at the run time while creating the Spark session as well.

Now, create the Spark session, and you are ready to play with Delta tables.

Environment Setup

How do you add the Delta Lake dependencies to your environment?

1. You can directly add them while initializing the Spark session for Delta Lake by passing the specific version, and these packages or dependencies will be downloaded during run time.
2. You can place the required .jar files in your cluster and provide the reference while initializing the Spark session.
3. You can download the .jar files and store them in cloud storage, and you can pass them as a run time argument if you don’t want to download the dependencies on your cluster.

# Initialize Spark Session
import pyspark
from delta import *
from pyspark.sql.types import *
from delta.tables import *
from pyspark.sql.functions import *

builder = pyspark.sql.SparkSession.builder.appName("My App") \
      .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \


         .config("spark.jars.packages", "io.delta:delta-core_2.12:2.2.0") \



# or if jar is already there

builder = pyspark.sql.SparkSession.builder.appName("My App") \
      .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")




spark = builder.getOrCreate()

# Initialize Spark Session
import pyspark
from delta import *
from pyspark.sql.types import *
from delta.tables import *
from pyspark.sql.functions import *

builder = pyspark.sql.SparkSession.builder.appName("My App") \
      .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \


         .config("spark.jars.packages", "io.delta:delta-core_2.12:2.2.0") \



# or if jar is already there

builder = pyspark.sql.SparkSession.builder.appName("My App") \
      .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")




spark = builder.getOrCreate()

You have to add the following properties to use delta in Spark:-

• Spark.sql.extensions
• Spark.sql.catalog.spark_catalog

You can see these values in the above code snippet. If you want to use cloud storage like reading and writing data from S3, GS, or Blob storage, then we have to set some more configs as well in the Spark session. Here, I am providing examples for AWS and GSC only.

The next thing that will come to your mind: how will you be able to read or write the data into cloud storage?

For different cloud storages, there are certain .jar files available that are used to connect and to do IO operations on the cloud storage. See the examples below.

You can use the above approach to make this .jar available for Spark sessions either by downloading at a run time or storing them on the cluster itself.

AWS 

spark_jars_packages = “com.amazonaws:aws-java-sdk:1.12.246,org.apache.hadoop:hadoop-aws:3.2.2,io.delta:delta-core_2.12:2.2.0”

spark = SparkSession.builder.appName(‘delta’)
  .config(“spark.jars.packages”, spark_jars_packages)
  .config(“spark.sql.extensions”, “io.delta.sql.DeltaSparkSessionExtension”)
  .config(“spark.sql.catalog.spark_catalog”, “org.apache.spark.sql.delta.catalog.DeltaCatalog”)
  .config(‘spark.hadoop.fs.s3a.aws.credentials.provider’, ‘org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider’)
  .config(“spark.hadoop.fs.s3.impl”, “org.apache.hadoop.fs.s3a.S3AFileSystem”)
  .config(“spark.hadoop.fs.AbstractFileSystem.s3.impl”, “org.apache.hadoop.fs.s3a.S3AFileSystem”)
  .config(“spark.delta.logStore.class”, “org.apache.spark.sql.delta.storage.S3SingleDriverLogStore”) 

spark = builder.getOrCreate()

‍

GCS

spark_session = SparkSession.builder.appName(‘delta’).builder.getOrCreate()

‍

spark_session.conf.set(“fs.gs.impl”, “com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem”)

spark_session.conf.set(“spark.hadoop.fs.gs.impl”, “com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem”)

spark_session.conf.set(“fs.gs.auth.service.account.enable”, “true”)

spark_session.conf.set(“fs.AbstractFileSystem.gs.impl”, “com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS”)

spark_session.conf.set(“fs.gs.project.id”, project_id)

spark_session.conf.set(“fs.gs.auth.service.account.email”, credential[“client_email”])

spark_session.conf.set(“fs.gs.auth.service.account.private.key.id”, credential[“private_key_id”])

spark_session.conf.set(“fs.gs.auth.service.account.private.key”, credential[“private_key”])‍

Write into Delta Tables: In the following example, we are using a local system only for reading and writing the data into and from delta lake tables.

Data Set Used: https://media.githubusercontent.com/media/datablist/sample-csv-files/main/files/organizations/organizations-100.zip

For reference, I have downloaded this file in my local machine and unzipped the data:

df = spark.read.option("header", "true").csv("organizations-100.csv")
df.write.mode('overwrite').format("delta").partitionBy(partition_keys).save("./Documents/DE/Delta/test-db/organisatuons")

df = spark.read.option("header", "true").csv("organizations-100.csv")
df.write.mode('overwrite').format("delta").partitionBy(partition_keys).save("./Documents/DE/Delta/test-db/organisatuons")

There are two modes available in Delta Lake and Spark (Append and Overwrite) while writing the data in the Delta tables from any source.

For now, we have enabled the Delta catalog to store all metadata-related information. We can also use the hive meta store to store the metadata information and to directly run the SQL queries over the delta tables. You can use the cloud storage path as well.

Read data from the Delta tables:

delta_df = spark.read.format("delta").load("./Documents/DE/Delta/test-db/organisatuons")
delta_df.show()

delta_df = spark.read.format("delta").load("./Documents/DE/Delta/test-db/organisatuons")
delta_df.show()

‍

Here, you can see the folder structure, and after writing the data into Delta Tables, it creates one delta log file, which keeps track of metadata, partitions, and files.

Option 2: Create Delta Table and insert data using Spark SQL.

spark.sql("CREATE TABLE orgs_data(index String, c_name String, organization_id String, name String, website String, country String, description String, founded String, industry String, num_of_employees String, remarks String) USING DELTA")

spark.sql("CREATE TABLE orgs_data(index String, c_name String, organization_id String, name String, website String, country String, description String, founded String, industry String, num_of_employees String, remarks String) USING DELTA")

Insert the data:

df.write.mode('append').format("delta").option("mergeSchema", "true").saveAsTable("orgs_data")
spark.sql("SELECT * FROM orgs_data").show()
spark.sql("DESCRIBE TABLE orgs_data").show()

df.write.mode('append').format("delta").option("mergeSchema", "true").saveAsTable("orgs_data")
spark.sql("SELECT * FROM orgs_data").show()
spark.sql("DESCRIBE TABLE orgs_data").show()

This way, we can read the Delta table, and you can use SQL as well if you have enabled the hive.

Schema Enforcement: Safeguarding Your Data

In the realm of data management, maintaining the integrity of your dataset is paramount. Delta Lake, with its schema enforcement capabilities, ensures that your data is not just welcomed with open arms but also closely scrutinized for compatibility. Let’s dive into the meticulous checks 

Delta Lake performs when validating incoming data against the existing schema:

Column Presence: Delta Lake checks that every column in your DataFrame matches the columns in the target Delta table. If there’s a single mismatch, it won’t let the data in and, instead, will raise a flag in the form of an exception.

Data Types Harmony: Data types are the secret language of your dataset. Delta Lake insists that the data types in your incoming DataFrame align harmoniously with those in the target Delta table. Any discord in data types will result in a raised exception.

Name Consistency: In the world of data, names matter. Delta Lake meticulously examines that the column names in your incoming DataFrame are an exact match to those in the target Delta table. No aliases allowed. Any discrepancies will lead to, you guessed it, an exception.

This meticulous schema validation guarantees that your incoming data seamlessly integrates with the target Delta table. If any aspect of your data doesn’t meet these strict criteria, it won’t find a home in the Delta Lake, and you’ll be greeted by an error message and a raised exception.

Schema Evolution: Adapting to Changing Data

In the dynamic landscape of data, change is the only constant. Delta Lake’s schema evolution comes to the rescue when you need to adapt your table’s schema to accommodate incoming data. This powerful feature offers two distinct approaches:

Overwrite Schema: You can choose to boldly overwrite the existing schema with the schema of your incoming data. This is an excellent option when your data’s structure undergoes significant changes. Just set the “overwriteSchema” option to true, and voila, your table is reborn with the new schema.

Merge Schema: In some cases, you might want to embrace the new while preserving the old. Delta Lake’s “Merge Schema” property lets you merge the incoming data’s schema with the existing one. This means that if an extra column appears in your data, it elegantly melds into the target table without throwing any schema-related tantrums.

Should you find the need to tweak column names or data types to better align with the incoming data, Delta Lake’s got you covered. The schema evolution capabilities ensure your dataset stays in tune with the ever-changing data landscape. It’s a smooth transition, no hiccups, and no surprises, just data management at its finest.

spark.read.table(...) 
  .withColumn("birthDate", col("birthDate").cast("date")) 
  .write 
  .format("delta") 
  .mode("overwrite")
  .option("overwriteSchema", "true") 
  .saveAsTable(...)

spark.read.table(...) 
  .withColumn("birthDate", col("birthDate").cast("date")) 
  .write 
  .format("delta") 
  .mode("overwrite")
  .option("overwriteSchema", "true") 
  .saveAsTable(...)

The above code will overwrite the existing delta table with the new schema along with the new data.

Delta Lake has support for automatic schema evolution. For instance, if you have added two more columns in the Delta Lake tables and still tried to access the existing table, you will be able to read the data without any error.

There is another way as well. For example, if you have three columns in a Delta table but the incoming table has four columns, you can set up spark.databricks.delta.schema.autoMerge.enabled is true. It can be done for the entire cluster as well.

spark.sql("DESCRIBE TABLE orgs_data").show()

spark.sql("DESCRIBE TABLE orgs_data").show()

Let’s add one more column and try to access the data again:

spark.sql(“alter table orgs_data add column(exta_col String)”)

spark.sql(“describe table orgs_data”).show()

As you can see, that column has been added but has not impacted the data. You can still smoothly and seamlessly read the data. It will set null to a newly created column.

What happens if we receive the extra column in an incoming CSV that we want to append to the existing delta table? You have to set up one config here for that:

input_df = spark.read.format('csv').option('header', 'true').load("../Desktop/Data-Engineering/data-samples/input-data/organizations-11111.csv")
input_df.printSchema()
input_df.write.mode('append').format("delta").option("mergeSchema", "true").saveAsTable("orgs_data")
spark.sql("SELECT * FROM orgs_data").show()
spark.sql("DESCRIBE TABLE orgs_data").show()

input_df = spark.read.format('csv').option('header', 'true').load("../Desktop/Data-Engineering/data-samples/input-data/organizations-11111.csv")
input_df.printSchema()
input_df.write.mode('append').format("delta").option("mergeSchema", "true").saveAsTable("orgs_data")
spark.sql("SELECT * FROM orgs_data").show()
spark.sql("DESCRIBE TABLE orgs_data").show()

You have to add this config mergeSchem=true while appending the data. It will merge the schema of incoming data that is receiving some extra columns.

The first figure shows the schema of incoming data, and in the previous one, we have already seen the schema of our delta tables.

Here, we can see that the new column that was coming in the incoming data is merged with the existing schema of the table. Now, the delta table has the latest updated schema.

Time Travel 

Basically, Delta Lake keeps track of all the changes in _delta_log by creating a log file. By using this, we can fetch the data of the previous version by specifying the version number.

df = spark.read.format("delta").option("versionAsOf", 0).load("orgs_data")
df.show()

df = spark.read.format("delta").option("versionAsOf", 0).load("orgs_data")
df.show()

Here, we can see the first version of the data, where we have not added any columns. As we know, the Delta table maintains the delta log file, which contains the information of each commit so that we can fetch the data till the particular commit.

Upsert, Delete, and Merge

Unlocking the Power of Upsert with Delta Lake

In the exhilarating realm of data management, upserting shines as a vital operation, allowing you to seamlessly merge new data with your existing dataset. It’s the magic wand that updates, inserts, or even deletes records based on their status in the incoming data. However, for this enchanting process to work its wonders, you need a key—a primary key, to be precise. This key acts as the linchpin for merging data, much like a conductor orchestrating a symphony.

A Missing Piece: Copy on Write and Merge on Read

Now, before we delve into the mystical world of upserting with Delta Lake, it’s worth noting that Delta Lake dances to its own tune. Unlike some other table formats like Hudi and Iceberg, Delta Lake doesn’t rely on the concepts of Copy on Write and Merge on Read. These techniques are used elsewhere to speed up data operations.

Two Paths to Merge: SQL and Spark API

To harness the power of upserting in Delta Lake, you have two pathways at your disposal: SQL and Spark API. The choice largely depends on your Delta version. In the latest Delta version, 2.2.0, you can seamlessly execute merge operations using Spark API. It’s a breeze. However, if you’re working with an earlier Delta version, say 1.0.0, then Spark SQL is your trusty steed for upserts and merges. Remember, using the right Delta version is crucial, or you might find yourself grappling with the cryptic “Method not found” error, which can turn into a debugging labyrinth.

In the snippet below, we showcase the elegance of upserting using Spark SQL, a technique that ensures your data management journey is smooth and error-free:

-- Insert new data and update existing data based on the specified key
MERGE INTO targetTable AS target
USING sourceTable AS source
ON target.id = source.id
WHEN MATCHED THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *
WHEN NOT MATCHED BY SOURCE THEN
  DELETE;

-- Insert new data and update existing data based on the specified key
MERGE INTO targetTable AS target
USING sourceTable AS source
ON target.id = source.id
WHEN MATCHED THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *
WHEN NOT MATCHED BY SOURCE THEN
  DELETE;

today_data_df = spark.read.format('csv').option('header', 'true').load("../Desktop/Data-Engineering/data-samples/input-data/organizations-11111.csv")
today_data_df.show()
spark.sql("select * from orgs_data where organization_id = 'FAB0d41d5b5ddd'").show()
# Reading Existing Delta table
deltaTable = DeltaTable.forPath(spark, "orgs_data")
today_data_df.createOrReplaceTempView("incoming_data")

today_data_df = spark.read.format('csv').option('header', 'true').load("../Desktop/Data-Engineering/data-samples/input-data/organizations-11111.csv")
today_data_df.show()


spark.sql("select * from orgs_data where organization_id = 'FAB0d41d5b5ddd'").show()


# Reading Existing Delta table
deltaTable = DeltaTable.forPath(spark, "orgs_data")

today_data_df.createOrReplaceTempView("incoming_data")

Here, we are loading the incoming data and showing what is inside. The existing data with the same primary key appears in the Delta table so that we can compare after upserting or merging the data.

spark.sql(
"""
MERGE INTO orgs_data
USING incoming_data
ON orgs_data.organization_id = incoming_data.organization_id
WHEN MATCHED THEN
  UPDATE SET
    organization_id = incoming_data.organization_id,
    name = incoming_data.name
"""
)
spark.sql("select * from orgs_data where organization_id = 'FAB0d41d5b5ddd'").show()

spark.sql(
"""
MERGE INTO orgs_data
USING incoming_data
ON orgs_data.organization_id = incoming_data.organization_id
WHEN MATCHED THEN
  UPDATE SET
    organization_id = incoming_data.organization_id,
    name = incoming_data.name
"""
)

spark.sql("select * from orgs_data where organization_id = 'FAB0d41d5b5ddd'").show()

‍

orgs_data.alias("oldData").merge(
   incoming_data.alias("newData"),
   f"oldData.organization_id = newData.organization_id") 
   .whenMatchedUpdateAll() 
   .whenNotMatchedInsertAll() 
   .execute()

orgs_data.alias("oldData").merge(
   incoming_data.alias("newData"),
   f"oldData.organization_id = newData.organization_id") 
   .whenMatchedUpdateAll() 
   .whenNotMatchedInsertAll() 
   .execute()

This is the example of how you can do upsert using Spark APIs. The merge operation creates lots of small files. You can control the number of small files by setting up the following properties in the spark session.

spark.delta.merge.repartitionBeforeWrite true

spark.sql.shuffle.partitions 10

This is how merge operations work. Merge supports one-to-one mapping. What we want to say is that only rows should try to update the one row in the target delta table. If multiple rows try to update the rows in the target Delta table, it will fail. Delta Lake matches the data on the basis of a key in case of an update operation.

Change Data Feed

This is also a useful feature of Delta Lake and tracks and maintains the history of all records in the Delta table after upsert or insert at the row level. You can enable these things at the beginning while setting up the Spark session or using Spark SQL by enabling  “change events” for all the data.

Now, you can see the whole journey of each record in the Delta table, from its insertion to deletion. It introduces one more extra column, _change_type, which contains the type of operations that have been performed on that particular row.

To enable this, you can set these configurations: 

spark.sql("set spark.databricks.delta.properties.defaults.enableChangeDataFeed = true;") 

spark.sql("set spark.databricks.delta.properties.defaults.enableChangeDataFeed = true;") 

Or you can set this conf while reading the delta table as well. 

## Stream Data Generation
data = [{"Category": 'A', "ID": 1, "Value": 121.44, "Truth": True, "Year": 2022},
        {"Category": 'B', "ID": 2, "Value": 300.01, "Truth": False, "Year": 2020},
        {"Category": 'C', "ID": 3, "Value": 10.99, "Truth": None, "Year": 2022},
        {"Category": 'E', "ID": 5, "Value": 33.87, "Truth": True, "Year": 2022}
        ]
df = spark.createDataFrame(data)
df.show()
df.write.mode('overwrite').format("delta").partitionBy("Year").save("silver_table")

## Stream Data Generation

data = [{"Category": 'A', "ID": 1, "Value": 121.44, "Truth": True, "Year": 2022},
        {"Category": 'B', "ID": 2, "Value": 300.01, "Truth": False, "Year": 2020},
        {"Category": 'C', "ID": 3, "Value": 10.99, "Truth": None, "Year": 2022},
        {"Category": 'E', "ID": 5, "Value": 33.87, "Truth": True, "Year": 2022}
        ]

df = spark.createDataFrame(data)

df.show()

df.write.mode('overwrite').format("delta").partitionBy("Year").save("silver_table")

deltaTable = DeltaTable.forPath(spark, "silver_table")
                                 
deltaTable.delete(condition = "ID == 1")
delta_df = spark.read.format("delta").option("readChangeFeed", "true").option("startingVersion", 0).load("silver_table")
delta_df.show()

deltaTable = DeltaTable.forPath(spark, "silver_table")
                                 
deltaTable.delete(condition = "ID == 1")

delta_df = spark.read.format("delta").option("readChangeFeed", "true").option("startingVersion", 0).load("silver_table")
delta_df.show()

Now, after deleting something, you will be able to see the changes, like what is deleted and what is updated. If you are doing upserts on the same Delta table after enabling the change data feed, you will be able to see the update as well, and if you insert anything, you will be able to see what is inserted in your Delta table. 

If we overwrite the complete Delta table, it will mark all past records as a delete:

If you want to record each data change, you have to enable this before creating the table so that we can see the data changes for each version. If you’ve already created one table, you won’t be able to see the changes for the previous version once you enable the change data feed, but you will be able to see the changes in all versions that came after this configuration.

Data Skipping with Z-ordering

Data skipping is the technique in Delta Lake where, if you have a larger number of records stored in many files, it will read the data from the files that contain required information, but apart from that, other files will get skipped. This makes it faster to read the data from the Delta tables.

Z-ordering is a technique used to colocate the information in the same dataset files. If you know the column that will be more in use in the select statement and has A cardinality, you can use Z-order by that particular column. It will reduce the large number of files from being read. We can give you multiple columns for Z-order by separating them from commas.

Let’s suppose you have two tables, a and b, and there is one column that is most frequently used. You can increase the number of files to be skipped, and you can use the columns files running that query. Normal order works linearly, whereas Z-order works in multi- dimensionally.

OPTIMIZE events
WHERE date >= current_timestamp() - INTERVAL 1 day
ZORDER BY (eventType)

OPTIMIZE events
WHERE date >= current_timestamp() - INTERVAL 1 day
ZORDER BY (eventType)

DML Operations

Delta Lake has capabilities to run all the DML operations of SQL in the data lake as well as update, delete, and merge operations.

Integrations and Ecosystem Supported in Delta Lake

‍Read Delta Tables

Unlock the Delta Tables: Tools That Bring Data to Life

Reading data from Delta tables is like diving into a treasure trove of information, and there’s more than one way to unlock its secrets. Beyond the standard Spark API, we have a squad of powerful allies ready to assist: SQL query engines like Athena and Trino. But they’re not just passive onlookers; they bring their own magic to the table, empowering you to perform data manipulation language (DML) operations that can reshape your data universe.

Athena: Unleash the SQL Sorcery

Imagine Athena as the Oracle of data. With SQL as its spellbook, it delves deep into your Delta tables, fetching insights with precision and grace. But here’s the twist: Athena isn’t just for querying; like a skilled blacksmith, it can help you hammer your data into a new shape, creating a masterpiece.

Trino: The Shape-Shifting Wizard

Trino, on the other hand, is the shape-shifter of the data realm. It glides through Delta tables, allowing you to perform an array of DML operations that can transform your data into new, dazzling forms. Think of it as a master craftsman who can sculpt your data, creating entirely new narratives and visualizations.

So, when it comes to Delta tables, these tools are not just readers; they are your co-creators. They enable you to not only glimpse the data’s beauty but also mold it into whatever shape serves your purpose. With Athena and Trino at your side, the possibilities are as boundless as your imagination.

Read Delta Tables Using Spark APIS

from delta.tables import *
delta_df =DeltaTable.forPath(spark,"./Documents/DE/Delta/test-db/organisatuons")<br>delta_df.toDf().show()

from delta.tables import *
delta_df =DeltaTable.forPath(spark,"./Documents/DE/Delta/test-db/organisatuons")<br>delta_df.toDf().show()

Steps to Set Up Delta Lake with S3 on EC2 Or EMR and Access Data through Athena

Data Set Used – We have generated some dummy data of around 100gb and written that into the Delta tables.

Step 1- Set up a Spark session along with AWS cloud storage and Delta – Spark. Here, we have used an EC2 instance with Spark 3.3 and Delta version 2.1.1. Here, we are setting up Spark config for Delta and S3.

AWS_ACCESS_KEY_ID = "XXXXXXXXXXXXXXXXXXXXXX"
AWS_SECRET_ACCESS_KEY = "XXXXXXXXXXXXXXXXXXXXXX+XXXXXXXXXXXXXXXXXXXXXX"

spark_jars_packages = "com.amazonaws:aws-java-sdk:1.12.246,org.apache.hadoop:hadoop-aws:3.2.2,io.delta:delta-core_2.12:2.1.1"

spark = SparkSession.builder.appName('delta') \
   .config("spark.jars.packages", spark_jars_packages) \
   .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
   .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
   .config('spark.hadoop.fs.s3a.aws.credentials.provider', 'org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider') \
   .config("spark.hadoop.fs.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
   .config("spark.hadoop.fs.AbstractFileSystem.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
   .config("spark.delta.logStore.class", "org.apache.spark.sql.delta.storage.S3SingleDriverLogStore") \
   .config("spark.driver.memory", "20g") \
   .config("spark.memory.offHeap.enabled", "true") \
   .config("spark.memory.offHeap.size", "8g") \
   .getOrCreate()

spark.sparkContext._jsc.hadoopConfiguration().set("fs.s3a.access.key", AWS_ACCESS_KEY_ID)
spark.sparkContext._jsc.hadoopConfiguration().set("fs.s3a.secret.key", AWS_SECRET_ACCESS_KEY)

AWS_ACCESS_KEY_ID = "XXXXXXXXXXXXXXXXXXXXXX"
AWS_SECRET_ACCESS_KEY = "XXXXXXXXXXXXXXXXXXXXXX+XXXXXXXXXXXXXXXXXXXXXX"

spark_jars_packages = "com.amazonaws:aws-java-sdk:1.12.246,org.apache.hadoop:hadoop-aws:3.2.2,io.delta:delta-core_2.12:2.1.1"

spark = SparkSession.builder.appName('delta') \
   .config("spark.jars.packages", spark_jars_packages) \
   .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
   .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
   .config('spark.hadoop.fs.s3a.aws.credentials.provider', 'org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider') \
   .config("spark.hadoop.fs.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
   .config("spark.hadoop.fs.AbstractFileSystem.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
   .config("spark.delta.logStore.class", "org.apache.spark.sql.delta.storage.S3SingleDriverLogStore") \
   .config("spark.driver.memory", "20g") \
   .config("spark.memory.offHeap.enabled", "true") \
   .config("spark.memory.offHeap.size", "8g") \
   .getOrCreate()

spark.sparkContext._jsc.hadoopConfiguration().set("fs.s3a.access.key", AWS_ACCESS_KEY_ID)
spark.sparkContext._jsc.hadoopConfiguration().set("fs.s3a.secret.key", AWS_SECRET_ACCESS_KEY)

Spark Version – You can use any Spark version, but Spark 3.3.1 came along with the pip install. Just make sure whatever version you are using is compatible with the Delta Lake version that you are using; otherwise, most of the features won’t work.

Step 2 – Here, we are creating a Delta table with an S3 path. We can directly write the data into an S3 bucket as a Delta table, but it is better to create a table first and then write it into S3 to make sure the schema is correct.

Set the Delta location path if it exists to run the Spark SQL query and create the Delta table along with the S3 path. 

# If table is already there
delta_path = "s3a://abhishek-test-01012023/delta-lake-sample-data/"
spark.conf.set('table.location', delta_path)

# Creating new delta table on s3 location
spark.sql("CREATE TABLE delta.`s3://abhishek-test-01012023/delta-lake-sample-data/`(id INT, first_name String, "
         "last_name String, address String, pincocde INT, net_income INT, source_of_income String, state String, "
         "email_id String, description String, population INT, population_1 String, population_2 String, "
         "population_3 String, population_4 String, population_5 String, population_6 String, population_7 String, "
         "date String) USING DELTA PARTITIONED BY (date)")

# If table is already there
delta_path = "s3a://abhishek-test-01012023/delta-lake-sample-data/"
spark.conf.set('table.location', delta_path)

# Creating new delta table on s3 location
spark.sql("CREATE TABLE delta.`s3://abhishek-test-01012023/delta-lake-sample-data/`(id INT, first_name String, "
         "last_name String, address String, pincocde INT, net_income INT, source_of_income String, state String, "
         "email_id String, description String, population INT, population_1 String, population_2 String, "
         "population_3 String, population_4 String, population_5 String, population_6 String, population_7 String, "
         "date String) USING DELTA PARTITIONED BY (date)")

Step 3 – Here, I have given one link that I have used to generate the dummy data and have written that into the S3 bucket as Delta tables. Feel free to look over this. An example code of writing is given below:

df.write.format("delta").mode("append").partitionBy("date").save("s3a://abhishek-test-01012023/delta-lake-sample-data/")

df.write.format("delta").mode("append").partitionBy("date").save("s3a://abhishek-test-01012023/delta-lake-sample-data/")

https://github.com/velotio-tech/delta-lake-iceberg-poc/blob/0396cdbf96230609695a907fdbe8c240042fce9e/delta-data-writer.py#L83

In the above link, you find the code of dummy data generation.

Step 4 – Here, we are printing the count and selecting some data from the Delta table that we have written in just right away.

Run the SQL query to check the table data and upsert using S3 bucket data:

spark.sql("select count(*) from delta.`s3://abhishek-test-01012023/delta-lake-sample-data/` group by id having count("
         "*) > 1").show()

spark.sql("select count(*) from delta.`s3://abhishek-test-01012023/delta-lake-sample-data/`").show()

#############################################################################
# Upsert
#############################################################################

# upserts the starting five records. We will read first five record and will do some changes in some columns and

input_df = spark.read.csv("s3a://abhishek-test-01012023/incoming_data/delta/4e0ae9f5-8c9d-435a-a434-febff1effbc3.csv",inferSchema=True,header=True)
input_df.printSchema()
input_df.createOrReplaceTempView("incoming")
spark.sql("MERGE INTO delta.`s3://abhishek-test-01012023/delta-lake-sample-data/` t USING (SELECT * FROM incoming) s ON t.id = s.id WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *")

spark.sql("select count(*) from delta.`s3://abhishek-test-01012023/delta-lake-sample-data/` group by id having count("
         "*) > 1").show()

spark.sql("select count(*) from delta.`s3://abhishek-test-01012023/delta-lake-sample-data/`").show()

#############################################################################
# Upsert
#############################################################################

# upserts the starting five records. We will read first five record and will do some changes in some columns and

input_df = spark.read.csv("s3a://abhishek-test-01012023/incoming_data/delta/4e0ae9f5-8c9d-435a-a434-febff1effbc3.csv",inferSchema=True,header=True)
input_df.printSchema()
input_df.createOrReplaceTempView("incoming")
spark.sql("MERGE INTO delta.`s3://abhishek-test-01012023/delta-lake-sample-data/` t USING (SELECT * FROM incoming) s ON t.id = s.id WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *")

This is the output of select statement:

This is the schema of the incoming data we are planning to merge into the existing Delta table:

After upsert, let’s see the data for the particular data partition:

spark.sql(“select * from delta.`s3://abhishek-test-01012023/delta-lake-sample-data/` where id = 1 and date = 20221206”)

‍

‍

Access Delta table using Hive or any other external metastore: 

For that, we have to create a link between them and to create this link, go to the Spark code and generate the manifest file on the S3 path where we have already written the data.

spark.sql("GENERATE symlink_format_manifest FOR TABLE 
delta.`s3a://abhishek-test-01012023/delta-lake-sample-data/`")

spark.sql("GENERATE symlink_format_manifest FOR TABLE 
delta.`s3a://abhishek-test-01012023/delta-lake-sample-data/`")

This will create the manifest folder not go to Athena and run this query:

CREATE EXTERNAL TABLE delta_db.delta_table(id INT, first_name String, last_name String, address String, pincocde INT, net_income INT, source_of_income String, state String, email_id String, description String, population INT, population_1 String, population_2 String, population_3 String, population_4 String, population_5 String, population_6 String, population_7 String) 
PARTITIONED BY (date String)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://abhishek-test-01012023/delta-lake-sample-data/_symlink_format_manifest/'

CREATE EXTERNAL TABLE delta_db.delta_table(id INT, first_name String, last_name String, address String, pincocde INT, net_income INT, source_of_income String, state String, email_id String, description String, population INT, population_1 String, population_2 String, population_3 String, population_4 String, population_5 String, population_6 String, population_7 String) 
PARTITIONED BY (date String)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://abhishek-test-01012023/delta-lake-sample-data/_symlink_format_manifest/'

MSCK REPAIR TABLE delta_db.delta_table

MSCK REPAIR TABLE delta_db.delta_table

You will be able to query the data. 

Conclusion: A New Dawn

In a world where data continued to grow in volume and complexity, Delta Lake stood as a beacon of hope. It empowered organizations to manage their data lakes with unprecedented efficiency and extract insights with unwavering confidence.

The adoption of Delta Lake marked a new dawn in the realm of data. Whether dealing with structured or semi-structured data, it was the answer to the prayers of data warriors. As the sun set on traditional formats, Delta Lake emerged as the hero they had been waiting for—a hero who had not only revolutionized data storage and processing but also transformed the way stories were told in the world of data.

And so, the legend of Delta Lake continued to unfold, inspiring data adventurers across the land to embark on their own quests, armed with the power of ACID transactions, time travel, and the promise of a brighter, data-driven future.

Introduction

Have you ever wondered how Google Maps, Starlink, Zomato, Arogya Setu, and even methods like population clustering are able to add value to the human world? Well, the common thread between these applications and technologies is the use of spatial data and analysis techniques.

Both Google Maps and Zomato use spatial techniques to provide navigation and location-based information to their users. While Arogya Setu is a contact tracing app that uses spatial data to track the spread of infectious illnesses, Starlink uses spatial data analysis to provide internet access to remote areas around the world. Population clustering is a technique that can be useful for urban planning, public health, and disaster response. Since the use of spatial data and its analysis techniques has become increasingly critical in the current scenario, let’s understand some fundamentals and explore different aspects of spatial data analytics.

So, welcome to the world of spatial data analytics, where data meets geography and insights come to life! The use of spatial data analytics has changed the way we understand and interact with the world around us, providing insights and solutions to some of the most pressing challenges facing humanity today. So, let’s cut through the process by taking a quick tour of a spatial journey that you might have never been on before.

What is spatial data analytics?

Before we start talking about the process of spatial data analytics, let’s try to understand what is special about the term “spatial data.” Spatial data, also known as “Geospatial data,” refers to data representing features or objects on the Earth’s surface. Whether it’s man-made or natural, if it has to do with a specific location on the surface of the Earth, it’s spatial. Spatial data refers to where things are now, or perhaps, where they were or will be in the future.

This data can be further classified as:

Geometric Data:

Geometric data is a type of spatial data mapped on a two-dimensional flat surface. Google Maps is an application that uses geometric data to provide accurate directions.

Geographic Data:

Geographic data is information mapped around the Earth that highlights the latitude and longitude relationships to a specific object or location. A familiar example of geographic data is a Global Positioning System (GPS).

Spatial data is not limited to structured information; it also comprises imagery from satellites and drones, address data points, and longitudinal and latitudinal data. Primarily, spatial data is classified as vector data and raster data. Vector data consists of coordinate information, while raster data is all about layers of images extracted from camera sensors.

The real world can be represented as below, where the built environment (roads, buildings) and administrative data (countries, census areas) tend to be represented as vector data. Natural environment (e.g., elevation, temperature, precipitation) is often represented using a raster grid.

1. Discrete data, stored according to its exact geographic location, is called vector data.
2. Continuous data is represented by regular grids called raster data.
3. Attributes

Vector Data:

• Points: A single dot on the layer depicts them. It can be either an x, y, or z coordinate.
• Lines: This form of vector data is presented using two coordinates, i.e., either the x-y coordinate or the inverse of this, and has a definite length. These are used for rivers, roads, railways, ferry routes, and even major pipeline flows.
• Polygon: The feature is defined using three or more coordinates. It is used to showcase inland water bodies like lakes, buildings, etc.

Raster Data:

• Raster is all about multilayered map images from satellites, drones, and various other camera sensors (ortho-imagery).
• It is stored in cell-based and color-pixel formats. These pixels are arranged in columns and rows.
• Analysis can be done better than with vector-based data due to the richness of the data.
• It can give you more accurate measurements than other types of data.

Attributes and Properties:

• Spatial data contains more information than just a location on the surface of the Earth.
• Any additional information, or non-spatial data, that describes a feature is referred to as an attribute.
• In addition to locational and attribute information, spatial data inherently contains geometric and topological properties, which help to gain deeper insights.
• Geometric properties include position and measurements, such as length, direction, area, and volume.
• Topological properties represent spatial relationships such as connectivity, inclusion, and adjacency.

As seen above, spatial data includes information such as geographic coordinates, elevation, and demographic information. Hence, it can be used to identify patterns, correlations, and trends that are not readily apparent through other data sources. For instance, geospatial data can be used to map the distribution of air pollution across a city, identify areas at risk for natural disasters like floods or wildfires, or monitor changes in land use over time. Here is where the analytics process takes place to uncover insights that can aid in providing solutions.

Spatial data analytics involves collecting, processing, and analyzing various types of spatial data with insights to go beyond what occurs to determine not only where and when something occurs but also why it occurs at that specific place and/or time. It can be further viewed as descriptive analytics, which involves summarizing and visualizing spatial data to identify patterns and trends. Predictive analytics uses statistical models to make predictions about future events or trends based on past data. Prescriptive analytics uses optimization techniques to determine the best course of action given a specific set of circumstances.

Why is spatial data analytics important?

Spatial data analytics plays an essential role in many industries and fields, providing insights and solutions that can have a significant impact on our daily lives. It aids businesses in gaining a competitive edge through improved decision-making and time and money savings. Urban planning, telecommunications, military, public health, and emergency management are just a few examples of industries that rely heavily on spatial data analytics to make informed decisions.

‍Public Health

A patient’s location directly influences their health. Whether it’s disease prevention or clinic site selection, considering spatial aspects in healthcare analytics can have a drastic impact.

Urban Planning

An urban planner might want to assess the extent of urban fringe growth, quantify the population growth that some suburbs are witnessing, and also understand why these particular suburbs are growing, and others are not.

Environmental and Natural Resources

Protecting our world against climate change, promoting biodiversity, exploration, and conservation planning requires spatial storytelling and sophisticated environmental analysis.

Space and Navigation

Optimizing transport infrastructure and navigation spatially is key to the future of mobility. The most efficient cities are moving away from traditional methods to analyze new data. 

Telecommunication

Since network signal strength fluctuates by location over time, spatial analytics helps telecommunications companies understand where anomalies occur and then resolve them.

Architecture, Engineering, and Construction

The leading AEC firms are going beyond traditional workflows to use spatial data science in urban planning and site selection, reducing costs and boosting project profitability. A geological engineer might want to identify the best localities for constructing buildings in an earthquake-prone area by looking at rock formation characteristics.

Military

Spatial predictive analytics helps the military optimize the placement of resources while using predictive analytics to assess infrastructure, situational awareness, anticipate maintenance needs, and meet deadlines.

Weather Forecasting

Rapid response to extreme weather by visualizing blizzards, wildfires, and hurricanes fast enough for effective evacuation alerts. Spatial data analytics also helps airlines with routing and gives insurance companies a better way to assess property risk.

How to perform spatial data analytics?

The process of spatial data analytics involves data gathering, data cleaning, data processing, and visualization, much like any other traditional analytics technique. The specific details of the process will be determined on the basis of the data and the goals of the analysis.

‍

‍Data Collection: The initial stage in spatial data analytics is to collect the relevant data. This involves gathering data from different sources, such as remote sensing satellites, GPS-enabled devices, social media, or survey instruments. The data may include geographic coordinates, attributes of features, and other pertinent information that can help analyze the data.

Data Cleaning and Preprocessing: Once the data is collected, it needs to be cleaned and preprocessed to ensure that it is accurate and usable for further processing. This may involve eliminating duplicates, filling in missing values, and standardizing data formats.

Data Transformation: Spatial data is often obtained from numerous sources and in a variety of forms, so the next step is to transform and combine the data into a single data set. This may involve joining tables or layers based on a shared attribute or location.

Data Analysis: This part of spatial data analytics involves identifying spatial patterns and relationships in the data. This may involve various techniques such as clustering, interpolation, spatial regression, and spatial autocorrelation analysis. The analysis may also include visualizing the data using maps, charts, and graphs for spatial data exploration.

Modeling and Prediction: Based on the results of spatial analysis, it may be possible to build models to predict future patterns or trends in the data as a part of predictive analytics. This may involve using machine learning algorithms or other statistical techniques to identify patterns and make predictions.

Business Intelligence: Finally, the results of spatial data analytics can be used to support decision-making in a variety of contexts, such as urban planning, natural resource management, or emergency response. The decision-making process may involve evaluating trade-offs between different options and considering the potential impact of different decisions on the spatial patterns in the data.

Tools and Techniques:

Spatial Data Storage

Spatial data storage is a specialized form of data storage that takes into account the spatial relationships between various data points, allowing for more efficient and effective analysis and retrieval of information. There are many tools available for spatial data storage, including both open-source and proprietary software. Here are a few instances of such tools.

‍

‍RDBMS (Relational Database Management Systems): RDBMS are among the most used methods for storing geographical data having extensions that enable spatial features. RDBMS examples supporting geographic data include:

• PostgreSQL with PostGIS extension
• Oracle Spatial
• Microsoft SQL Server with SQL Server Spatial

Spatial File Formats: Spatial file formats are widely used for storing and sharing spatial data. Examples of spatial file formats include:

• Shapefile (.shp)
• GeoJSON (.geojson)
• Keyhole Markup Language (KML) (.kml)
• Geography Markup Language (GML) (.gml)

NoSQL: NoSQL databases are becoming increasingly popular for spatial data storage due to their ability to handle large and complex datasets, flexible schema, and scalability. Examples of NoSQL databases that support spatial data include:

• MongoDB with GeoJSON support
• Cassandra with GeoMesa extension
• Elasticsearch with GeoPoint data type

Cloud-based Storage Services: Cloud-based storage services like AWS, GCP, Azure are popular options for storing spatial data, which can be termed as DataLakes. Examples of cloud-based storage services that support spatial data include:

• Amazon S3 with Amazon S3 GeoSpatial Indexing
• Google Cloud Storage with Google Cloud Storage Geo-Location
• Microsoft Azure Blob Storage with Azure Spatial Anchors

Spatial Data Warehouses: Spatial data warehouses are specialized databases designed for spatial data analysis. Examples of spatial data warehouses include:

• ArcGIS Online
• CARTO
• GeoServer

It can be noted that tools, such as RDBMS and NoSQL databases, can also be used for spatial data analytics and processing in addition to storage.

Spatial Data Processing

Spatial data processing is an important step in spatial data analytics to ensure that the data is properly aligned and in a consistent format before further analysis. This is a must-do step because various applications and data sources use different formats and coordinate systems, which might lead to several difficulties when analyzing the data.

Below are a few examples of processing methods in a spatial context that ensure that spatial data is compatible and consistent across different applications and data sources.

Reprojection: Reprojection is the process of converting spatial data from one map projection to another. This is frequently necessary when working with data from multiple sources that use different projections.

Coordinate System/Datum Transformation: This transformation involves converting spatial data from one coordinate system to another or from one geodetic datum to another. This is important when working with data from different sources that use different coordinate systems and information.

Resampling: Resampling involves changing the resolution or scale of spatial data. This is often necessary when handling data at different scales coming from different sources.

Geocoding: Geocoding is the process of converting a street address or other location description into a set of geographic coordinates. This allows the location to be plotted on a map and later analyzed in a spatial context.

Georeferencing: Georeferencing is the method of aligning geographic data to a specific coordinate system or reference system. This is often required when working with data from several sources, such as aerial photographs or satellite imagery.

Digitizing: Digitizing is the process of converting analog maps or other spatial data into a digital format. This involves manually tracing features such as roads, buildings, and water bodies using a computer program.

Several tools are available that can perform such data processing techniques, and a few of these tool instances are given below.

GIS (Geographic Information Systems): GIS connects data to a map, integrating location data with all types of descriptive information. It helps users understand patterns, relationships, and geographic context. The benefits include improved communication and efficiency, as well as better management and decision-making. Examples of GIS software that supports spatial data processing include:

• ArcGIS – A proprietary GIS software with a comprehensive set of features and tools
• QGIS – An open-source GIS software with a wide range of plugins and tools

Python Libraries: Python is a popular programming language for spatial data processing, and there are several libraries available for this purpose. Examples of Python libraries that support spatial data processing include:

• GeoPandas: A library for working with geospatial data in Python
• Shapely: A library for manipulation and analysis of planar geometric objects
• PySAL: A library for spatial analysis and modeling

R Packages: Like Python, R is another popular programming language for spatial data processing, and there are several packages available for spatial data operations. Examples of R packages that support spatial data processing include:

• sf: an R package for working with geospatial data
• sp: an R package for spatial data analysis
• raster: an R package for working with raster data

SQL: SQL can be used for spatial data processing and analysis, especially when working with spatial databases with extensions like PostGIS. Examples of SQL spatial functions include:

• ST_Contains: checks if one geometry contains another
• ST_Intersects: checks if two geometries intersect
• ST_Distance: calculates the distance between two geometries

Command-line Tools: There are a handful of command-line tools available for spatial data processing. Examples of command-line tools that support spatial data processing include:

• GDAL/OGR: a suite of tools for geospatial data processing and conversion
• GRASS GIS: a command-line tool for geospatial analysis and modeling

There are a few other tools for data processing worth exploring, such as MATLAB, GeoServer, Global Mapper, and Mapbox. Tools like GIS software and Python libraries can also be used for spatial data storage and analysis in addition to processing.

‍Spatial Data Analysis

Spatial data analysis is the process of examining geographic data to spot trends, correlations, and patterns. It involves the use of statistical, computational, and visualization methods to explore spatial data and extract business insights. There are different categories of spatial data analysis, such as:

Proximity Analysis: It involves measuring the distance between two or more places in a spatial dataset. It is possible to analyze proximity using methods like Euclidean distance.

Accessibility Analysis: It is a measure of how easy it is to get to a location from other locations in the dataset. In addition to distance, Accessibility analysis takes into account other factors that affect how easy it is to travel between locations, such as traffic, road conditions, and public transportation.

Spatial Clustering: Spatial clustering is the process of identifying groups of spatially adjacent objects that have similar characteristics. Hierarchical clustering and k-means clustering are two methods that can be used to accomplish this.

Spatial Interpolation: Spatial interpolation involves estimating values for locations where data is not available based on nearby data points. This can be done using techniques such as kriging or inverse distance weighting.

Spatial Exploratory Data Analysis: It involves creating visual representations of spatial data to explore patterns and relationships. Spatial EDA helps to identify patterns and relationships that may not be immediately apparent from the data and can help guide further analysis. This can include techniques such as choropleth maps, heat maps, or scatter plots.

Spatial Simulation: This involves using simulation models to study the behavior of spatial systems over time. Spatial simulation includes techniques such as cellular automata, agent-based models, and Monte Carlo simulations. Spatial simulation is useful for predicting the future behavior of spatial systems under different scenarios.

There are other categories, like factor analysis, trajectory analysis, network analysis, etc., that can be used for fine-grained spatial data analysis. Below are a few examples of tools that can be used to devise an analysis of spatial data.

GIS (Geographic Information Systems): As seen earlier, this software can be used not only for capturing and processing the data but also to analyze and display geographically referenced data. Examples include ArcGIS, QGIS, and GRASS GIS.

Open Source Libraries and Binaries: It includes various programming languages with many packages for spatial data analysis, such as the sp package for handling spatial data and the rgdal package for reading and writing geospatial data formats in R or packages, such as GeoPandas and Shapely to provide functionality for working with geospatial data in Python. The list can go on with the GDAL framework and its dependencies.

PostGIS: PostGIS is a spatial database extender for the PostgreSQL database management system. It adds support for geographic objects, allowing you to manipulate and query geospatial data within the database for any kind of analysis purpose.

Data Visualization Tools: These tools are used to create visual representations of spatial data for exploratory data analysis. Examples include Tableau, ArcGIS Pro, and QGIS.

Mapbox: Mapbox is a mapping platform that provides APIs and SDKs for building custom maps and applications. It includes tools for data visualization, geocoding, routing, and more.

ENVI: ENVI is a software package for processing and analyzing remote sensing data. It includes tools for image classification, spectral analysis, and terrain modeling, among others.

Spatial data analysis plays an essential role in understanding complex spatial patterns and relationships and can help formulate business decisions in a wide range of areas. The choice of tool depends on the type of analysis that needs to be performed, the size of the data set, and the resources available.

How to solve spatial big data problems?

Big data refers to datasets that are too large and complex to process and analyze using the traditional methods that we discussed earlier. When dealing with spatial data, the challenges of big data are amplified due to the added dimensions of space and time. Spatial data is being captured at an unusual rate because of the growing numbers of sensors and devices, the networks of GPS satellites and cell towers, and the rise of the Internet of Things.

Spatial data analytics can leverage strategies and resources, including distributed computing, cloud computing, and parallel processing, to address the above issues. These techniques allow for the processing and analysis of large spatial datasets, enabling real-time decision-making in industries including transportation, agriculture, and public safety. For instance, massive geographic data analytics are used by real-time traffic management systems to optimize traffic flows, ease congestion, and improve safety.

‍Apache Sedona (formerly GeoSpark):

• Apache Spark with a geospatial extension for geospatial data analytics capabilities.
• Supports different spatial indexes, such as R-Tree, Quadtree, and K-D Tree, which can improve the performance of spatial queries and operations.
• Supports various spatial queries, such as range queries, KNN queries, and spatial joins.
• Designed to work with other components of the Spark ecosystem, such as Spark SQL and Spark Streaming.
• Provides support for machine learning algorithms on geospatial data, such as clustering and classification.

SpatialHadoop:

• Apache Hadoop with a geospatial extension for spatial data analytics capabilities.
• Can process and analyze large-scale spatial data in a distributed environment using the MapReduce paradigm.
• Supports different spatial indexes, such as R-Tree and Grid File, which can improve the performance of spatial queries and operations.
• Supports various spatial queries, such as range queries, KNN queries, and spatial joins.
• Designed to work with other components of the Hadoop ecosystem, such as HDFS, MapReduce, and Hive.

BigQuery GIS:

• Google Cloud Platform that provides geospatial data analytics capabilities.
• It is a fully-managed service that automatically scales up or down based on the volume of data and the complexity of queries.
• Supports different spatial indexes, such as R-Tree and Hilbert Curve, which can improve the performance of spatial queries and operations.
• Supports various spatial queries, such as range queries, KNN queries, and spatial joins.
• Designed to work with other components of the BigQuery ecosystem, such as BigQuery ML, BigQuery BI Engine, and Bigquery Geo Viz.

There are a few other tools and extensions, like Esri GIS Tools for Hadoop, SpatialSpark, Google Earth Engine that can be used to gain insights and make informed decisions based on spatial data.

Case Study

In the telecommunications industry, spatial analysis can be used to optimize network coverage and capacity, plan new infrastructure, and identify areas of high network congestion.

Let’s consider a hypothetical telecommunications company that wants to improve its network performance and customer experience by analyzing geospatial data. Specifically, the company wants to analyze call detail records (CDRs) to identify areas of high call volume and network congestion.

In a given solution by Azure in a published article, the suggested architecture involves:

• Azure Data Factory, which is used to collect the CDRs from various sources (mainly geospatial databases).
• Azure Data Factory stores them in Azure Data Lake Storage in formats such as GeoJSON, WKT, and Vector tiles. The bronze container holds raw data, the silver container holds semi-curated data, and the gold container holds fully curated data as the processing proceeds.
• Azure Databricks and the GeoSpark/Sedona package are being used to convert data formats and efficiently load, process, and analyze large-scale spatial data across machines.
• GeoPandas exports data in various formats, which are later used by GIS applications such as QGIS and ArcGIS for exploratory analysis.
• Azure Machine Learning extracts insights from geospatial data, determining, for example, where and when to deploy new wireless access points.
• Power BI or Azure Maps can be used to visualize the geospatial data and identify areas where network upgrades or infrastructure improvements are needed.
• A log analytics system is set up to run queries against data in Azure Monitor Logs to implement a robust and fine-grained logging system to analyze events and performance.

Overall, the Azure-based solution gives an idea about how one can try to perform geospatial analysis in the telecommunications industry and improve network performance and customer experience. You can read more about this solution here.

Challenges and limitations

In conclusion, spatial data analytics is an essential component of decision-making across a range of industries. It is important to understand the techniques, infrastructure, and challenges of spatial data analytics to effectively leverage spatial data and make informed decisions. Spatial data collection can be challenging and may contain faults or inconsistencies. The data may not be available for certain geographic areas or for certain time periods. Spatial data analytics can raise privacy concerns if personal data is collected and used without consent. In addition, there may be concerns about the use of spatial data analytics for surveillance or other unethical purposes, which can lead to significant harm.

Conclusion

Spatial data analytics is a powerful tool that can help organizations make better-informed decisions and gain a competitive advantage. As the fields of machine learning (AI) and spatial data analysis intertwine, spatial data analytics looks promising and quite useful for real-life problems. The blend of both vector and raster data produces a powerful product that can tackle various economic and earth-related problems. This blog is just a high-level overview of spatial data analytics since you have just scratched the surface, but I can guarantee that this spatial ride will be smoother from here on.

As marketed, Spinnaker is an open-source, multi-cloud continuous delivery platform that helps you release software changes with high velocity and confidence.

Open sourced by Netflix and heavily contributed to by Google, it supports all major cloud vendors (AWS, Azure, App Engine, Openstack, etc.) including Kubernetes.

In this blog I’m going to walk you through all the basic concepts in Spinnaker and help you create a continuous delivery pipeline using Kubernetes Engine, Cloud Source Repositories, Container Builder, Resource Manager, and Spinnaker. After creating a sample application, we will configure these services to automatically build, test, and deploy it. When the application code is modified, the changes trigger the continuous delivery pipeline to automatically rebuild, retest, and redeploy the new version.

What Spinnaker Provides?

Application management and Application Deployment are its two core features.

Application Management

Spinnaker’s application management features can be used to view and manage your cloud resources.

Modern tech organizations operate collections of services—sometimes referred to as “applications” or “microservices”. A Spinnaker application models this concept.

Applications, Clusters, and Server Groups are the key concepts Spinnaker uses to describe services. Load balancers and Firewalls describe how services are exposed to users.

Application

• An application in Spinnaker is a collection of clusters, which in turn are collections of server groups. The application also includes firewalls and load balancers. An application represents the service which needs to be deployed using Spinnaker, all configuration for that service, and all the infrastructure on which it will run. Normally, a different application is configured for each service, though Spinnaker does not enforce that.

Cluster

• Clusters are logical groupings of Server Groups in Spinnaker.
• Note: Cluster, here, does not map to a Kubernetes cluster. It’s merely a collection of Server Groups, irrespective of any Kubernetes clusters that might be included in your underlying architecture.

Server Group

• The base resource, the Server Group, identifies the deployable artifact (VM image, Docker image, source location) and basic configuration settings such as number of instances, autoscaling policies, metadata, etc. This resource is optionally associated with a Load Balancer and a Firewall. When deployed, a Server Group is a collection of instances of the running software (VM instances, Kubernetes pods).

Load Balancer

• A Load Balancer is associated with an ingress protocol and port range. Traffic is balanced among the instances present in Server Groups. Optionally, health checks can be enabled for a load balancer, with flexibility to define health criteria and specify the health check endpoint.

Firewall

• A Firewall defines network traffic access. It is effectively a set of firewall rules defined by an IP range (CIDR) along with a communication protocol (e.g., TCP) and port range.

Application Deployment

Pipeline

• The pipeline is the key deployment management construct in Spinnaker. It consists of a sequence of actions, known as stages. Parameters can be passed from one stage to the next one in the pipeline.
• You can start a pipeline manually, or you can configure it to be automatically triggered by an event, such as a Jenkins job completing, a new Docker image being pushed in your docker registry, a CRON type schedule, or maybe a stage in another pipeline.
• You can configure the pipeline to emit notifications, by email, SMS or HipChat, to interested parties at various points during pipeline execution (such as on pipeline start/complete/fail).

Stage

• A Stage in Spinnaker is an atomic building block for a pipeline, describing an action that the pipeline will perform. You can sequence stages in a Pipeline in any order, though some stage sequences may be more common than others. There are different types of stages in Spinnaker such as Deploy, Manual Judgment, Resize, Disable,  and many more. The full list of stages and read about implementation details for each provider here.

Deployment Strategies

• Spinnaker supports all the cloud native deployment strategies including Red/Black (a.k.a Blue/Green), Rolling red/black and Canary deployments, etc.

What is Spinnaker Made Of?

Spinnaker is composed of a number of independent microservices:

• Deck Deck is the custom browser-based GUI.
• Gate is the API gateway. All the API calls from UI (Deck) and other API callers go to Spinnaker through Gate.
• Orca is the orchestration engine. It handles all ad-hoc operations and pipelines.
• Clouddriver is responsible for all mutating calls to the cloud providers and for indexing/caching all deployed resources.
• Front50 is used to persist the metadata of applications, pipelines, projects and notifications.
• Rosco is the bakery. It helps to create machine images for various cloud vendors (for example GCE images for GCP, AMIs for AWS, Azure VM images). It currently wraps Packer, but will be expanded to support additional mechanisms for producing images.
• Igor is used to trigger pipelines via continuous integration jobs in systems like Jenkins and Travis CI, and it allows Jenkins/Travis stages to be used in pipelines.
• Echo is Spinnaker’s eventing bus. It supports sending notifications (e.g. Slack, email, Hipchat, SMS), and acts on incoming webhooks from services like GitHub.
• Fiat is Spinnaker’s authorization service. It is used to query a user’s access permissions for accounts, applications and service accounts.
• Kayenta provides automated canary analysis for Spinnaker.
• Halyard is Spinnaker’s configuration service. Halyard manages the lifecycle of each of the above services. It only interacts with these services during Spinnaker start-up, updates, and rollbacks.

By default, Spinnaker binds ports accordingly for all the above mentioned microservices. For us the UI (Deck) will be exposed onto Port 9000.

What are We Going to Do?

• Set up your environment by launching Cloud Shell, creating a Kubernetes Engine cluster, and configuring your identity and user management scheme.
• Download a sample application, create a Git repository, and upload it to a Cloud Source Repository.
• Deploy Spinnaker to Kubernetes Engine using Helm.
• Build a Docker image from the source code.
• Create triggers to create Docker images when the source code for application changes.
• Configure a Spinnaker pipeline to reliably and continuously deploy your application to Kubernetes Engine.
• Deploy a code change, triggering the pipeline, and watch it roll out to production.

 Note: This blog post uses various billable components in GCP like GKE, Container Builder etc. 

Pipeline Architecture

To continuously deliver application updates to users, companies need an automated process that reliably builds, tests, and updates their software. Code changes should automatically flow through a pipeline that includes artifact creation, unit testing, functional testing, and production rollout. In some cases, they want a code update to apply to only a subset of their users, so that it is exercised realistically before pushing it to entire user base. If one of these canary releases proves unsatisfactory, the automated procedure must be able to quickly roll back the software changes.

With Kubernetes Engine and Spinnaker, we can create a robust continuous delivery flow that helps us to ensure that software is shipped as quickly as it is developed and validated. Although rapid iteration is the end goal, we must first ensure that each application revision passes through a series of automated validations before becoming a candidate for production rollout. When a given change has been vetted through automation, we can also validate the application manually and conduct further pre-release testing.

After the team decides the application is ready for production, one of the team members can approve it for production deployment.

Application Delivery Pipeline

We are going to build the continuous delivery pipeline shown in the following diagram.

Prerequisites  

• Fair bit of experience in GCP services like:  
• GKE (Google Kubernetes Engine)
• Google Compute
• Google APIs
• Cloud Source Repository
• Container Builder
• Cloud Storage
• Cloud Load Balancing
• Knowledge in K8s terminology like Services, Deployments, Pods, etc
• Familiarity with Kubectl and Helm package manager

Before Starting just enable the APIs needed on GCP

• Kubernetes API
• Compute API
• Resource Manger API
• IAM API

 Set Up a Kubernetes Cluster  

1. Go to the Console and scroll the left panel down to Compute->Kubernetes Engine->Kubernetes Clusters.
2. Click Create Cluster.
3. Choose a name or leave as the default one.
4. Under Machine Type, click Customize.
5. Allocate at least 2 vCPU and 10GB of RAM.
6. Change the cluster size to 2.
7. Enable Legacy Authorization while customizing the cluster.
8. Keep the rest of the defaults and click Create.

In a minute or two the cluster will be created and ready to go.

Configure identity and access management

Create a Cloud Identity and Access Management (Cloud IAM) service account to delegate permissions to Spinnaker, allowing it to store data in Cloud Storage. Spinnaker stores its pipeline data in Cloud Storage to ensure reliability and resiliency. If our Spinnaker deployment unexpectedly fails, we can create an identical deployment in minutes with access to the same pipeline data as the original.

1. Create the service account:

$ gcloud iam service-accounts create spinnaker-storage-account  --display-name spinnaker-storage-account

$ gcloud iam service-accounts create spinnaker-storage-account  --display-name spinnaker-storage-account

2.  Store the service account email address and our current project ID in environment variables for use in later commands:

$ export SA_EMAIL=$(gcloud iam service-accounts list  --filter="displayName:spinnaker-storage-account"  --format='value(email)')
$ export PROJECT=$(gcloud info --format='value(config.project)')

$ export SA_EMAIL=$(gcloud iam service-accounts list  --filter="displayName:spinnaker-storage-account"  --format='value(email)')
$ export PROJECT=$(gcloud info --format='value(config.project)')

3. Bind the storage.admin role to our service account:  

$ gcloud projects add-iam-policy-binding  $PROJECT --role roles/storage.admin --member serviceAccount:$SA_EMAI

$ gcloud projects add-iam-policy-binding  $PROJECT --role roles/storage.admin --member serviceAccount:$SA_EMAI

4. Download the service account key. We will need this key later while installing Spinnaker and we need to also upload the key to Kubernetes Engine.  

$ gcloud iam service-accounts keys create spinnaker-sa.json --iam-account $SA_EMAIL

$ gcloud iam service-accounts keys create spinnaker-sa.json --iam-account $SA_EMAIL

Deploying Spinnaker using Helm

In this section, we will deploy Spinnaker onto the K8s cluster via Charts with the help of K8s package manager Helm. Helm has made it very easy to deploy Spinnaker, it can be a very painful act to deploy it manually via Halyard and configure it.

Install Helm

1. Download and install the helm binary:

$ wget https://storage.googleapis.com/kubernetes-helm/helm-v2.9.0-linux-amd64.tar.gz

$ wget https://storage.googleapis.com/kubernetes-helm/helm-v2.9.0-linux-amd64.tar.gz

2. Unzip the file to your local system:

$ tar zxfv helm-v2.9.0-linux-amd64.tar.gz$ sudo chmod +x linux-amd64/helm && sudo mv linux-amd64/helm /usr/bin/helm

$ tar zxfv helm-v2.9.0-linux-amd64.tar.gz$ sudo chmod +x linux-amd64/helm && sudo mv linux-amd64/helm /usr/bin/helm

3. Grant Tiller, the server side of Helm, the cluster-admin role in your cluster:

$ kubectl create clusterrolebinding user-admin-binding  --clusterrole=cluster-admin --user=$(gcloud config get-value account)
$ kubectl create serviceaccount tiller --namespace kube-system
$ kubectl create clusterrolebinding tiller-admin-binding  --clusterrole=cluster-admin --serviceaccount=kube-system:tiller

$ kubectl create clusterrolebinding user-admin-binding  --clusterrole=cluster-admin --user=$(gcloud config get-value account)
$ kubectl create serviceaccount tiller --namespace kube-system
$ kubectl create clusterrolebinding tiller-admin-binding  --clusterrole=cluster-admin --serviceaccount=kube-system:tiller

4. Grant Spinnaker the cluster-admin role so it can deploy resources across all namespaces:

$ kubectl create clusterrolebinding --clusterrole=cluster-admin       --serviceaccount=default:default spinnaker-admin

$ kubectl create clusterrolebinding --clusterrole=cluster-admin       --serviceaccount=default:default spinnaker-admin

5. Initialize Helm to install Tiller in your cluster:

$ helm init --service-account=tiller --upgrade
$ helm repo update

$ helm init --service-account=tiller --upgrade
$ helm repo update

6. Ensure that Helm is properly installed by running the following command. If Helm is correctly installed, v2.9.0 appears for both client and server.

$ helm version

$ helm version

Configure Spinnaker

1. Create a bucket for Spinnaker to store its pipeline configuration:

$ export PROJECT=$(gcloud info --format='value(config.project)')
$ export BUCKET=$PROJECT-spinnaker-configgsutil mb -c regional -l us-central1  gs://$BUCKET

$ export PROJECT=$(gcloud info --format='value(config.project)')
$ export BUCKET=$PROJECT-spinnaker-configgsutil mb -c regional -l us-central1  gs://$BUCKET

2. Create the configuration file:

$ export SA_JSON=$(cat spinnaker-sa.json)
$ export PROJECT=$(gcloud info --format='value(config.project)')
$ export BUCKET=$PROJECT-spinnaker-config
$ cat > spinnaker-config.yaml <

$ export SA_JSON=$(cat spinnaker-sa.json)
$ export PROJECT=$(gcloud info --format='value(config.project)')
$ export BUCKET=$PROJECT-spinnaker-config
$ cat > spinnaker-config.yaml <

# Disable minio as the defaultminio:      
enabled: false 
# Configure your Docker registries here accounts:      
name: gcr       
address: https://gcr.io 
username: _json_key 
password: '$SA_JSON'
email: 1234@5678.com EOF

# Disable minio as the defaultminio:      
enabled: false 

# Configure your Docker registries here accounts:      
name: gcr       
address: https://gcr.io 
username: _json_key 
password: '$SA_JSON'
email: 1234@5678.com EOF

Deploy the Spinnaker chart

1. Use the Helm command-line interface to deploy the chart with the configuration set earlier. This command typically takes five to ten minutes to complete, so we will be providing a deploy timeout with ` — timeout`.

$ helm install -n cd stable/spinnaker -f spinnaker-config.yaml --timeout  600 --version 0.3.1

$ helm install -n cd stable/spinnaker -f spinnaker-config.yaml --timeout  600 --version 0.3.1

After the command completes, run the following command to set up port forwarding to the Spinnaker UI from Cloud Shell:

$ export DECK_POD=$(kubectl get pods --namespace default -l  "component=deck" -o jsonpath="{.items[0].metadata.name}")
$ kubectl port-forward --namespace default $DECK_POD 8080:9000  >> /dev/null &

$ export DECK_POD=$(kubectl get pods --namespace default -l  "component=deck" -o jsonpath="{.items[0].metadata.name}")
$ kubectl port-forward --namespace default $DECK_POD 8080:9000  >> /dev/null &

The above command exposes the Spinnaker UI onto the local machine that we’re using to run all the commands. We can use any port of our choosing instead of 8080 in above command. Now the UI can be opened onto the url http://localhost:8080.

Building the Docker image

In this section, we will configure Container Builder to detect changes to the application source code, if yes then build a Docker image, and then push it to Container Registry.

For this step we will use a sample app provided by the Google community  

Create your source code repository

1. Download the source code:

$ wget https://gke-spinnaker.storage.googleapis.com/sample-app.tgz

$ wget https://gke-spinnaker.storage.googleapis.com/sample-app.tgz

2. Unpack the source code:

$ tar xzfv sample-app.tgz

$ tar xzfv sample-app.tgz

3. Change directories to source code:

$ cd sample-app

$ cd sample-app

4. Set the username and email address for Git commits in this repository. Replace [EMAIL_ADDRESS] with Git email address, and replace [USERNAME] with Git username.  

$ git config --global user.email "[EMAIL_ADDRESS]"
$ git config --global user.name "[USERNAME]"

$ git config --global user.email "[EMAIL_ADDRESS]"
$ git config --global user.name "[USERNAME]"

5. Make the initial commit to source code repository:

$ git init
$ git add .
$ git commit -m "Initial commit"

$ git init
$ git add .
$ git commit -m "Initial commit"

6. Create a repository to host the code:

$ gcloud source repos create sample-app
$ git config credential.helper gcloud.sh

$ gcloud source repos create sample-app
$ git config credential.helper gcloud.sh

7. Add our newly created repository as remote:

$ export PROJECT=$(gcloud info --format='value(config.project)')
$ git remote add origin  https://source.developers.google.com/p/$PROJECT/r/sample-app

$ export PROJECT=$(gcloud info --format='value(config.project)')
$ git remote add origin  https://source.developers.google.com/p/$PROJECT/r/sample-app

8. Push the code to the new repository’s master branch:

$ git push origin master

$ git push origin master

9. Check that we can see our source code in the console.

Configuring the build triggers  

In this section, we configure Google Container Builder to build and push your Docker images every time we push Git tags to our source repository. Container Builder automatically checks out the source code, builds the Docker image from the Dockerfile in repository, and pushes that image to Container Registry.

1. In the GCP Console, click Build Triggers in the Container Registry section.
2. Select Cloud Source Repository and click Continue.
3. Select your newly created sample-app repository from the list, and click Continue.
4. Set the following trigger settings:
5. Name:sample-app-tags
6. Trigger type: Tag
7. Tag (regex): v.*
8. Build configuration: cloudbuild.yaml
9. cloudbuild.yaml location: /cloudbuild.yaml
10. Click Create trigger.

From now on, whenever we push a Git tag prefixed with the letter “v” to source code repository, Container Builder automatically builds and pushes our application as a Docker image to Container Registry.

Let’s build our first image:

Push the first image using the following steps:

1. Go to source code folder in Cloud Shell.

2. Create a Git tag:

$ git tag v1.0.0

$ git tag v1.0.0

3. Push the tag:  

$ git push --tags

$ git push --tags

4. In Container Registry, click Build History to check that the build has been triggered. If not, verify the trigger was configured properly in the previous section.

Configuring your deployment pipelines

Now that our images are building automatically, we need to deploy them to the Kubernetes cluster.

We deploy to a scaled-down environment for integration testing. After the integration tests pass, we must manually approve the changes to deploy the code to production services.

Create the application

1. In the Spinnaker UI, click Actions, then click Create Application.

2. In the New Application dialog, enter the following fields:

1. Name: sample
2. Owner Email: [your email address]

3. Click Create.

Create service load balancers

To avoid having to enter the information manually in the UI, use the Kubernetes command-line interface to create load balancers for the services. Alternatively, we can perform this operation in the Spinnaker UI.

On the local machine where the code resides, run the following command from the sample-app root directory:

$ kubectl apply -f k8s/services

$ kubectl apply -f k8s/services

Create the deployment pipeline

Now we create the continuous delivery pipeline. The pipeline is configured to detect when a Docker image with a tag prefixed with “v” has arrived in your Container Registry.

1. Create a new pipeline named say “Deploy”.

2. Go to the Config page for the pipeline that we just created and click Pipeline Actions -> Edit as JSON.

3. Change the directory to the source code directory and update the current pipeline-deploy.json at path spinnaker/pipeline-deploy.json according to our needs.

$ export PROJECT=$(gcloud info --format='value(config.project)')
$ sed s/PROJECT/$PROJECT/g spinnaker/pipeline-deploy.json > spinnaker/updated-pipeline-deploy.json

$ export PROJECT=$(gcloud info --format='value(config.project)')
$ sed s/PROJECT/$PROJECT/g spinnaker/pipeline-deploy.json > spinnaker/updated-pipeline-deploy.json

4. Now in the JSON editor just copy the whole file spinnaker/updated-pipeline-deploy.json.

5. Click on Update Pipeline and we should have an updated pipeline config now.

6. In the Spinnaker UI, click Pipelines on the top navigation bar.

7. Click Configure in the Deploy pipeline.

8. The continuous delivery pipeline configuration appears in the UI:

Running the pipeline manually

The configuration we just created contains a trigger to start the pipeline when a new Git tag containing the prefix “v” is pushed. Now we test the pipeline by running it manually.  

1. Return to the Pipelines page by clicking Pipelines.

2. Click Start Manual Execution.

3. Select the v1.0.0 tag from the Tag drop-down list, then click Run.

4. After the pipeline starts, click Details to see more information about the build’s progress. This section shows the status of the deployment pipeline and its steps. Steps in blue are currently running, green ones have completed successfully, and red ones have failed. Click a stage to see details about it.

5. After 3 to 5 minutes the integration test phase completes and the pipeline requires manual approval to continue the deployment.

6. Hover over the yellow “person” icon and click Continue.

7. Your rollout continues to the production frontend and backend deployments. It completes after a few minutes.

8. To view the app, click Load Balancers in the top right of the Spinnaker UI.

9. Scroll down the list of load balancers and click Default, under sample-frontend-prod.  

10. Scroll down the details pane on the right and copy application’s IP address by clicking the clipboard button on the Ingress IP.

11. Paste the address into the browser to view the production version of the application.

12. We have now manually triggered the pipeline to build, test, and deploy your application. 

Triggering the pipeline automatically via code changes

Now let’s test the pipeline end to end by making a code change, pushing a Git tag, and watching the pipeline run in response. By pushing a Git tag that starts with “v”, we trigger Container Builder to build a new Docker image and push it to Container Registry. Spinnaker detects that the new image tag begins with “v” and triggers a pipeline to deploy the image to canaries, run tests, and roll out the same image to all pods in the deployment.

1. Change the colour of the app from orange to blue: 

$ sed -i 's/orange/blue/g' cmd/gke-info/common-service.go

$ sed -i 's/orange/blue/g' cmd/gke-info/common-service.go

view rawcolor.js hosted with ❤ by GitHub

2. Tag your change and push it to the source code repository:

$ git commit -a -m "Change colour to blue"git tag v1.0.1git push --tags

$ git commit -a -m "Change colour to blue"git tag v1.0.1git push --tags

view rawtag_color.js hosted with ❤ by GitHub

3. See the new build appear in the Container Builder Build History. 

4. Click Pipelines to watch the pipeline start to deploy the image. 

5. Observe the canary deployments. When the deployment is paused, waiting to roll out to production, start refreshing the tab that contains our application. Nine of our backends are running the previous version of your application, while only one backend is running the canary. Now we should see the new, blue version of our application appear about every tenth time we refresh. 

6. After testing completes, return to the Spinnaker tab and approve the deployment. 

7. When the pipeline completes, application looks like the following screenshot. Note that the colour has changed to blue because of code change, and that the Version field now reads v1.0.1. 

8. We have now successfully rolled out your application to your entire production environment!!!!!! 

9. Optionally, we can roll back this change by reverting the previous commit. Rolling back adds a new tag (v1.0.2), and pushes the tag back through the same pipeline we used to deploy v1.0.1: 

$ git revert v1.0.1
$ git tag v1.0.2
$ git push --tags

$ git revert v1.0.1
$ git tag v1.0.2
$ git push --tags

view rawrevert.js hosted with ❤ by GitHub

Conclusion

Now that you know how to get Spinnaker up and running in a development environment, start using it already. In this blog, we have done everything from installing a K8s cluster on GCP to deploying an End to End Pipeline just like that in a production environment. Hope you found it helpful.

References

https://cloud.google.com/solutions/continuous-delivery-spinnaker-kubernetes-engine

In today’s world, a lot of data is being generated daily. To process data that is large and very complex, traditional tools can’t be used. Huge volumes of complex data is simply called Big Data. Converting this raw data into meaningful insights, organizations can make better decisions with their products. We need a dedicated tool to help this raw data to be converted into meaningful data or knowledge. Thankfully, there are certain tools that can help.

Hadoop is one of the most popular frameworks used to process and store Big Data. Hive, in turn, is a tool that is designed to be used alongside Hadoop. In the blog, we are going to discuss the different ways we can load semi-structured and unstructured data into Hive. We will also be discussing what Hive is and how it works. How does the performance of Hive differ from working with structured vs. semi-structured vs. unstructured data?

What is Hive?

Hive is a data warehousing infrastructure tool developed on top of the Hadoop Distributed File System(HDFS). Hive can be used on top of any DFS.) Hive uses Hive query language(HQL), which is very much similar to structured query language(SQL). If you are familiar with SQL, then it is much easier to get started with HQL.

It is used for data querying and analysis over a large amount of data distributed over the Hadoop Distributed File System(HDFS). Hive supports reading, writing, and managing a large amount of data that is residing in the Hadoop Distributed File System(HDFS). Hive is mostly used for structured data but in this blog, we will see how we can load unstructured data.

Initially, Hive was developed at Facebook(Meta), and later it became an open-source project of Apache Software Foundation.

How does Hive work?

Source – AnalyticsVidya

Hive was created to allow non-programmers familiar with SQL to work with large datasets, using an HQL interface that is similar to SQL interface. Traditional databases are designed for small or medium datasets and not large ones. But Hive uses a distributed file system and batch processing to process large datasets very efficiently.

Hive transforms HQL queries into one or more Map-Reduce jobs or Tez jobs, and then these jobs run on Hadoop’s scheduler, YARN. Basically, HQL is an abstraction over Map-Reduce programs. After the execution of the job/query, the resulting data is stored in HDFS.

What is SerDe in Hive?

SerDe is short for “Serializer and Deserializer” in Hive. It’s going to be an important topic for this blog. So, you should have a basic understanding of what SerDe is and how it works.

If not, don’t worry, first of all, we will understand what Serialization and Deserialization is. When an object is converted into a byte stream, it’s into a binary format so that the object can be transmitted over a network or written into persistent storage like HDFS. This process of converting data objects into byte streams is called Serialization.

Now, we can transmit data objects or write data objects into persistent storage. But how can we receive transmitted data over the network again in a meaningful way because we will not be able to understand binary data properly? So, the process of converting byte stream or binary data back into objects is called Deserialization.

In Hive, tables are converted into row objects and row objects are written into HDFS using a Built-in Hive Serializer. And these row objects are converted back into tables using a Built-in Hive Deserializer.

Built-in SerDes:

• Avro (Hive 0.9.1 and later)

• ORC (Hive 0.11 and later)

• RegEx

• Thrift

• Parquet (Hive 0.13 and later)

• CSV (Hive 0.14 and later)

• JsonSerDe (Hive 0.12 and later in hcatalog-core)

Now, suppose we have data in a format that Hive’s Built-in SerDe can’t process. In such a scenario, we can write our own custom SerDe. Here, we will discuss how the row is converted into a table and vice versa. But writing your own custom SerDe is a complicated and complex process.

There is another way: we can use RegexSerDe. RegexSerDe uses a regular expression to serialize and deserialize the data using regex. RegexSerDe extracts groups as columns. Here, group means regular expressions capturing groups. In a regular expression, capturing groups are a way to treat multiple characters as a single unit. Groups can be created using placing parentheses. For example, the regular expression “(velotio)” creates a single group containing the characters “v,” “e,” “l,” “o,” ”t,” “i,” and “o.”

This is just an overview of SerDe in Hive, but you can deep dive into SerDe. Also, the following image shows the flow of How Hive reads and writes records.

Source: Dummies.com

Types of data :

Big data can be classified in three ways

Structured data:

The data that can be organized into a well-defined structure is called Structured Data. Structured data can be easily stored, read, or transferred in the same defined structure. The best example of structured data is the table stored in Relational Databases. Tables have columns and rows that define a well-organized and fixed structure to data. Another example of structured data is an Excel file. An Excel file also has rows and columns that define a proper structure to data.

Source: O’Reilly

Semi-structured data:

The data that can not be organized into a fixed structure like a table but can be represented with properties such as tags, metadata, or other markers that separate data fields are called semi-structured data. Examples of semi-structured data are JSON and XML files. JSON files contain “key” and “values” pairs, where the key is a tag, and the value is actual data to be stored.

Source: Software Testing Help

Unstructured data:

The data which can not be organized into any structure is called unstructured data. The social media messages fall under the unstructured data category as they can not be organized into either a fixed structure like a table or even with tags or markers that will separate data fields. More examples of unstructured data are text files, multimedia content like images and videos.

Source: Fluxicon

Performance impact of working with structured vs, semi-structured vs, unstructured data 

Storage:

Structured data is always stored in RDBMS. Structured data have a high organization level among all three. Semi-structured data has no schema but has some properties or tags. Structured data have less organization level compared to structured data but higher organization level than unstructured data. While unstructured data has no schema, so it has the lowest organization level.

Data manipulation:

Data manipulation includes updating and deleting data. Consider an example where we want to update the name of a student using his roll number. Data manipulation in structured data is easy to perform as we have defined structure and we can manipulate specific records very easily. We can easily update the student’s name using his roll number in structured data. Whereas in unstructured data, there is no schema available, so it is not easy to manipulate data in unstructured data as compared to structured data.

Searching of data:

Searching for particular data in structured data is easy compared to searching for data in unstructured data. In unstructured data, we will need to go through all lines, and each word and searching of data in unstructured data will get complex. Searching data in semi-structured data is also easy as we just need to specify the key to get the data.

Scaling of data:

Scaling structured data is very hard. It can be scaled vertically by adding an existing machine’s RAM or CPU, but scaling it horizontally is hard to do. However, scaling semi-structured data and unstructured data is easy.

Data sets we are using:

1) Video_Games_5.json

This dataset contains product item reviews and metadata from Amazon, including 142.8 million audits spreading over May 1996 – July 2014. Reviews (ratings, text, helpfulness votes), item metadata (depictions, category information, price, brand, and image features), and links (also viewed/also bought graphs) are all included in this dataset. The dataset represents data in a semi-structured manner.

The following image shows one record of the entire dataset.

{
  "reviewerID": "A2HD75EMZR8QLN",
  "asin": "0700099867",
  "reviewerName": "123",
  "helpful": [
    8,
    12
  ],
  "reviewText": "Installing the game was a struggle (because of games for windows live bugs).Some championship races and cars can only be \"unlocked\" by buying them as an addon to the game. I paid nearly 30 dollars when the game was new. I don't like the idea that I have to keep paying to keep playing.I noticed no improvement in the physics or graphics compared to Dirt 2.I tossed it in the garbage and vowed never to buy another codemasters game. I'm really tired of arcade style rally/racing games anyway.I'll continue to get my fix from Richard Burns Rally, and you should to. :)http://www.amazon.com/Richard-Burns-Rally-PC/dp/B000C97156/ref=sr_1_1?ie=UTF8&qid;=1341886844&sr;=8-1&keywords;=richard+burns+rallyThank you for reading my review! If you enjoyed it, be sure to rate it as helpful.",
  "overall": 1,
  "summary": "Pay to unlock content? I don't think so.",
  "unixReviewTime": 1341792000,
  "reviewTime": "07 9, 2012"}

{
  "reviewerID": "A2HD75EMZR8QLN",
  "asin": "0700099867",
  "reviewerName": "123",
  "helpful": [
    8,
    12
  ],
  "reviewText": "Installing the game was a struggle (because of games for windows live bugs).Some championship races and cars can only be \"unlocked\" by buying them as an addon to the game. I paid nearly 30 dollars when the game was new. I don't like the idea that I have to keep paying to keep playing.I noticed no improvement in the physics or graphics compared to Dirt 2.I tossed it in the garbage and vowed never to buy another codemasters game. I'm really tired of arcade style rally/racing games anyway.I'll continue to get my fix from Richard Burns Rally, and you should to. :)http://www.amazon.com/Richard-Burns-Rally-PC/dp/B000C97156/ref=sr_1_1?ie=UTF8&qid;=1341886844&sr;=8-1&keywords;=richard+burns+rallyThank you for reading my review! If you enjoyed it, be sure to rate it as helpful.",
  "overall": 1,
  "summary": "Pay to unlock content? I don't think so.",
  "unixReviewTime": 1341792000,
  "reviewTime": "07 9, 2012"}

2) sparkLog.txt

Apache Spark (https://spark.apache.org) is a unified analytics engine for big data processing, with built-in modules for streaming, SQL, machine learning, and graph processing. Currently, Spark has been widely deployed in the industry.

The Dataset / Log Set was collected by aggregating logs from the Spark system in a lab at CUHK, which comprises a total of 32 machines. The logs are aggregated at the machine level. Logs are provided as-is without further modification or labeling, which involve both normal and abnormal application runs.

The dataset represents data in an un-structured manner.

17/06/09 20:10:40 INFO executor.CoarseGrainedExecutorBackend: Registered signal handlers for [TERM, HUP, INT]
17/06/09 20:10:40 INFO spark.SecurityManager: Changing view acls to: yarn,curi
17/06/09 20:10:40 INFO spark.SecurityManager: Changing modify acls to: yarn,curi
17/06/09 20:10:40 INFO spark.SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(yarn, curi); users with modify permissions: Set(yarn, curi)
17/06/09 20:10:41 INFO spark.SecurityManager: Changing view acls to: yarn,curi
17/06/09 20:10:41 INFO spark.SecurityManager: Changing modify acls to: yarn,curi
17/06/09 20:10:41 INFO spark.SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(yarn, curi); users with modify permissions: Set(yarn, curi)
17/06/09 20:10:41 INFO slf4j.Slf4jLogger: Slf4jLogger started
17/06/09 20:10:41 INFO Remoting: Starting remoting
17/06/09 20:10:41 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkExecutorActorSystem@mesos-slave-07:55904]

17/06/09 20:10:40 INFO executor.CoarseGrainedExecutorBackend: Registered signal handlers for [TERM, HUP, INT]
17/06/09 20:10:40 INFO spark.SecurityManager: Changing view acls to: yarn,curi
17/06/09 20:10:40 INFO spark.SecurityManager: Changing modify acls to: yarn,curi
17/06/09 20:10:40 INFO spark.SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(yarn, curi); users with modify permissions: Set(yarn, curi)
17/06/09 20:10:41 INFO spark.SecurityManager: Changing view acls to: yarn,curi
17/06/09 20:10:41 INFO spark.SecurityManager: Changing modify acls to: yarn,curi
17/06/09 20:10:41 INFO spark.SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(yarn, curi); users with modify permissions: Set(yarn, curi)
17/06/09 20:10:41 INFO slf4j.Slf4jLogger: Slf4jLogger started
17/06/09 20:10:41 INFO Remoting: Starting remoting
17/06/09 20:10:41 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkExecutorActorSystem@mesos-slave-07:55904]

Now, we will look into different ways of loading unstructured data into Hive.

How to load semi-structured data into Hive?

1) Using Spark

If you are aware of Spark, loading semi-structured data into the spark is very easy. Spark can read JSON files, XML files, and convert them into Spark DataFrame. In Spark, DataFrame is a distributed collection of data that is organized into columns and rows. It is logically similar to tables in relational databases.

Now, we have our semi-structured data in an organized way. We can now write this organized DataFrame into Hive as a table from Spark.

Below is the code to read the JSON file and write it as a table in Hive.

‍

from pyspark.sql import SparkSession

## creating sparkSession to get entrypoint to spark application
sparkSession = SparkSession\
 .builder\
 .appName('Write_table_to_hive')\
 .enableHiveSupport()\
 .getOrCreate()

## reading data from dataset "Video_Games_5.json"
GamesReviewDataFrame = sparkSession.read.format("json") \
          .format("json") \
          .option("path", "/home/velotio/Downloads/UnstructuredData/Video_Games_5.json")\
          .load()

## we can modify data the way we want to represent in table here 
GamesReviewDataFrame.show()

## writing dataframe "GamesReviewDataFrame" as a table in HIVE.
GamesReviewDataFrame.write.saveAsTable("GameReviewTable")

sparkSession.stop()

from pyspark.sql import SparkSession

## creating sparkSession to get entrypoint to spark application
sparkSession = SparkSession\
 .builder\
 .appName('Write_table_to_hive')\
 .enableHiveSupport()\
 .getOrCreate()

## reading data from dataset "Video_Games_5.json"
GamesReviewDataFrame = sparkSession.read.format("json") \
          .format("json") \
          .option("path", "/home/velotio/Downloads/UnstructuredData/Video_Games_5.json")\
          .load()

## we can modify data the way we want to represent in table here 
GamesReviewDataFrame.show()

## writing dataframe "GamesReviewDataFrame" as a table in HIVE.
GamesReviewDataFrame.write.saveAsTable("GameReviewTable")

sparkSession.stop()

Output for above code:

As you can see in the output, a few records of the DataFrame are displayed in an organized table.

2) Using built-in SerDe, JSON SerDe

Hive provides us with a few built-in SerDe. Using this built-in SerDe, we can load data into Hive. In our case, we have used the Video_Games_5.json file as a dataset for semi-structured data, which is a JSON file. So, we will be using built-in JsonSerDe to load Video_Games_5.json data into Hive. This JsonSerDe can be used to read data in JSON format.

We will need to add JsonSerDe.jar to Hive.

You can download JsonSerDe here.

1) Copy dataset Video_Games_5.json from the local file system to the docker container.

To load data into the Hive table, we need to copy the dataset Video_Games_5.json into HDFS. As we are running HDFS and Hive in the docker container, we will need to copy this dataset from the Local File System to the docker container.

docker cp /home/velotio/Downloads/UnstructuredData/Video_Games_5.json 0fde53f41006:/aniket

docker cp /home/velotio/Downloads/UnstructuredData/Video_Games_5.json 0fde53f41006:/aniket

2) Copy dataset Video_Games_5.json from a docker container to the HDFS file system.

ls
ls /aniket
hdfs dfs -ls /
hdfs dfs -ls /aniket
hdfs dfs -put /aniket/Video_Games_5.json /aniket
hdfs dfs -ls /aniket

ls
ls /aniket
hdfs dfs -ls /
hdfs dfs -ls /aniket
hdfs dfs -put /aniket/Video_Games_5.json /aniket
hdfs dfs -ls /aniket

3) Copy json-serde.jar from the local file system to the docker container

To use JsonSerDe, add the json-serde.jar file to Hive so that Hive can use it.

We store this json-serde.jar file to HDFS storage where our dataset is also present. As Hive is running on top of HDFS, we can access the HDFS path from Hive. But to store json-serde.jar on HDFS, the file needs to be present in the docker container. For that, we copy json-serde.jar to the docker container first.

docker cp /home/velotio/Downloads/json-serde-1.3.7.3.jar 0fde53f41006:/aniket

docker cp /home/velotio/Downloads/json-serde-1.3.7.3.jar 0fde53f41006:/aniket

4) Copy json-serde.jar from a docker container to the HDFS file system.

ls /aniket
hdfs dfs -ls /aniket
hdfs dfs -put /aniket/json-serde-1.3.7.3.jar /aniket
hdfs dfs -ls /aniket

ls /aniket
hdfs dfs -ls /aniket

hdfs dfs -put /aniket/json-serde-1.3.7.3.jar /aniket
hdfs dfs -ls /aniket

5) Add json-serde.jar file to Hive

ADD JAR hdfs:///aniket/json-serde-1.3.7.3.jar;

ADD JAR hdfs:///aniket/json-serde-1.3.7.3.jar;

6) Create Hive table GameReviews

To load data into Hive and define the structure of our data, we must create a table in Hive before loading the data. The table holds the data in an organized manner.

While creating the table, we are specifying “row format serde,” which tells Hive to use the provided SerDe for reading and writing Hive data.

create table GameReviews(
reviewerID string,
asin string,
reviewerName string,
helpful array<int>,
reviewText string,
overall int,
summary string,
unixReviewTime int,
reviewTime string
)
row format serde 'org.apache.hive.hcatalog.data.JsonSerDe'
stored as textfile;

create table GameReviews(
reviewerID string,
asin string,
reviewerName string,
helpful array<int>,
reviewText string,
overall int,
summary string,
unixReviewTime int,
reviewTime string
)
row format serde 'org.apache.hive.hcatalog.data.JsonSerDe'
stored as textfile;

‍

7) Load data from the Video_Games_5.json dataset into the table.

We are loading data from Video_Games_5.json into the Hive table. With the help of SerDe provided while creating a table, Hive will parse this data and load it into the table.

load data inpath '/aniket/Video_Games_5.json' into table GameReviews;

load data inpath '/aniket/Video_Games_5.json' into table GameReviews;

8) Check data from the table.

Just cross-check if the data is loaded properly into the table.

select reviewerID,asin,reviewerName,overall from GameReviews limit 10;

select reviewerID,asin,reviewerName,overall from GameReviews limit 10;

How to load unstructured data into Hive ?

1. Using Regex SerDe

For unstructured data, the built-in SerDe can’t work with excluded RegxSerDe. To load unstructured data into Hive, we can use RegexSerde. First of all, we will need to figure out what unstructured data is useful. After knowing what data is useful, we can extract data using pattern matching. For that, we can use regular expressions. With regular expressions, we will load unstructured data of the SparkLog.txt dataset into Hive.

In our case, we are going to use the following regular expression:

“([0-9]{2}/[0-9]{2}/[0-9]{2}) ([0-9]{2}:[0-9]{2}:[0-9]{2}) [a-zA-Z]* ([a-zA-Z0-9.]*): (.*)$”

“([0-9]{2}/[0-9]{2}/[0-9]{2})”: First group in Regular Expression matches date values.

“([0-9]{2}:[0-9]{2}:[0-9]{2})”: Second Group in Regular Expression matches timestamp values.

“[a-zA-Z]*”: This pattern matches any string with multiple occurrences of char a to z and A to Z; this pattern will be ignored in the Hive table as we are not collecting this pattern as a group.

“([a-zA-Z0-9.]*):”: Third group in regular expression matches with multiple occurrences of char a to z, A to Z, 0 to 9 and “.”

“(.*)$”: Fourth and last group matches with all characters in the remaining string.

1) Copy dataset sparkLog.txt from local file system to docker container.

To load data into Hive table, we need dataset sparkLog.txt into HDFS. As we are running HDFS and Hive in the docker container, we will need to copy this dataset from the Local File System to the docker container first.

docker cp /home/velotio/Downloads/UnstructuredData/sparkLog.txt 6d94029a1f34:/aniket

docker cp /home/velotio/Downloads/UnstructuredData/sparkLog.txt 6d94029a1f34:/aniket

2) Copy dataset sparkLog.txt from a docker container to HDFS file system.

lsnls /aniketnhdfs dfs -ls /nhdfs dfs -ls /aniketnhdfs dfs -put /aniket/sparkLog.txt /aniketnhdfs dfs -ls /aniket

lsnls /aniketnhdfs dfs -ls /nhdfs dfs -ls /aniketnhdfs dfs -put /aniket/sparkLog.txt /aniketnhdfs dfs -ls /aniket

3) Create a Hive table sparkLog.

To load data into Hive and define the structure to our data, we must create a table in Hive before loading the data. The table holds the data in an organized manner.

While creating the table, we are specifying “row format SerDe,” which tells Hive to use the provided SerDe for reading and writing Hive data. For RegexSerDe, we must specify serdeproperties: “input.regex” and “output.format.string.”

create table sparkLog 
(
    datedata  string
   ,time string
   ,component string
      ,action string
)
    row format serde 'org.apache.hadoop.hive.serde2.RegexSerDe'
    with serdeproperties 
	(
		"input.regex" = "([0-9]{2}/[0-9]{2}/[0-9]{2}) ([0-9]{2}:[0-9]{2}:[0-9]{2}) [a-zA-Z]* ([a-zA-Z0-9.]*): (.*)$",
		"output.format.string" = "%1$s %2$s %3$s %4$s"
	)
STORED AS TEXTFILE;

create table sparkLog 
(
    datedata  string
   ,time string
   ,component string
      ,action string
)
    row format serde 'org.apache.hadoop.hive.serde2.RegexSerDe'
    with serdeproperties 
	(
		"input.regex" = "([0-9]{2}/[0-9]{2}/[0-9]{2}) ([0-9]{2}:[0-9]{2}:[0-9]{2}) [a-zA-Z]* ([a-zA-Z0-9.]*): (.*)$",
		"output.format.string" = "%1$s %2$s %3$s %4$s"
	)
STORED AS TEXTFILE;

4) Load data from the sparkLog.txt dataset into the table.

We are loading data from sparkLog.txt into the Hive table. With the help of the SerDe provided while creating the table, Hive will parse this data and will load it into the table.

load data inpath '/aniket/sparkLog.txt’ into table sparkLog;

load data inpath '/aniket/sparkLog.txt’ into table sparkLog;

5) Check the data from the table.

Cross-check if the data is loaded properly into the table.

select * from sparkLog;

select * from sparkLog;

2) Using HQL functions

For unstructured data, we have already seen how to use RegexSerDe to load unstructured data into Hive. But what if I am not aware of regular expressions or can’t write complex regular expressions to match patterns in a string? There is another way to load unstructured data into Hive using some HQL user-defined functions.

What we need to do is create a dummy table and load unstructured data as it is into Hive in just one column in the table named “line.” We are loading unstructured data into a dummy Hive table column named as a line. The first record of the “line” column will contain the first line of DataSet, and the second record of the line column will contain the second line of DataSet. Like this, the entire Dataset will be loaded into a dummy table.

Now, using HQL user-defined functions on the dummy Hive table, we can write specific data to specific columns into the main table using the “insert into” statement. You should be able to extract the data that you want using HQL user-defined functions.

1) Copy dataset sparkLog.txt from local file system to docker container

To load data into Hive table, we need dataset sparkLog.txt into HDFS. As we are running HDFS and Hive in the docker container, we will need to copy this dataset from the local file system to the docker container.

docker cp /home/velotio/Downloads/UnstructuredData/sparkLog.txt 6d94029a1f34:/aniket

docker cp /home/velotio/Downloads/UnstructuredData/sparkLog.txt 6d94029a1f34:/aniket

2) Copy the dataset sparkLog.txt from a docker container to HDFS file system.

ls
ls /aniket
hdfs dfs -ls /
hdfs dfs -ls /aniket
hdfs dfs -put /aniket/sparkLog.txt /aniket
hdfs dfs -ls /aniket

ls
ls /aniket
hdfs dfs -ls /
hdfs dfs -ls /aniket
hdfs dfs -put /aniket/sparkLog.txt /aniket
hdfs dfs -ls /aniket

3) Create Hive table log

We are creating a dummy Hive table as a log. We are specifying “row format delimited lines terminated by ‘/n’,” which tells Hive to consider default value for fields delimiter and ‘/n’ for line delimiter.

create table if not exists log 
(
	line string
)
row format delimited
lines terminated by ‘n’
STORED AS TEXTFILE;

create table if not exists log 
(
	line string
)
row format delimited
lines terminated by ‘n’
STORED AS TEXTFILE;

4)   Load the data from sparkLog.txt dataset into the table log.

We are loading data from sparkLog.txt into Hive table log.

load data inpath '/aniket/sparkLog.txt’ into table log;

load data inpath '/aniket/sparkLog.txt’ into table log;

5) Create Hive table log sparkLog.

We are creating a Hive table sparkLog to keep our organized data. This organized data will be extracted from a dummy Hive table log.

create table sparkLog 
(
   datedata  string
   ,time string
   ,component string
      ,action string
)
row format delimited
lines terminated by ‘\n’
STORED AS TEXTFILE;

create table sparkLog 
(
   datedata  string
   ,time string
   ,component string
      ,action string
)
row format delimited
lines terminated by ‘\n’
STORED AS TEXTFILE;

6) Parse the data from log table using a case statement and insert records into the sparkLog table.

We are using HQL user-defined functions to get the specific data and inserting this data into our sparkLog table using insert into statement.

insert into sparkLog select
split(line, ' ')[0] as datedata,
split(line, ' ')[1] as timedata,
split(split(line, ': ')[0],' ')[3] as component,
split(line, ': ')[1] as action
from log ;

insert into sparkLog select
split(line, ' ')[0] as datedata,
split(line, ' ')[1] as timedata,
split(split(line, ': ')[0],' ')[3] as component,
split(line, ': ')[1] as action
from log ;

7) Check data from the table.

Crosscheck if the data is loaded properly into the table.

select * from sparkLog;

select * from sparkLog;

Summary

After going through the above blog, you might have gotten more familiarity with Hive, its architecture. how you can use different serializers and deserializers in Hive. Now, you are able to load not only structured data but also unstructured data into Hive. If you are interested in knowing more about Apache Hive, you can visit the below documentation.

1. Hive Tutorial
2. LanguageManual
3. Hive Wiki Pages