Choice Between Sqoop_Hive and Spark for Data Ingestion

The choice between creating a data ingestion pipeline using Sqoop and Hive or using Spark depends on several factors, including your specific requirements and the characteristics of the source system. Here are some considerations for each option:

Creating a Pipeline with Sqoop and Hive:

Sqoop is a tool designed for efficiently transferring data between Hadoop (or HDFS) and relational databases. Hive is a data warehousing and SQL-like query language tool for Hadoop.

Pros:

Sqoop is well-suited for transferring data from traditional relational databases (e.g., MySQL, Oracle)       to Hadoop/Hive.

       1: It can handle large volumes of data efficiently and provides parallel processing capabilities.

       2: Hive provides a SQL-like interface for querying and analyzing data once it's ingested.

Cons:

1: Sqoop and Hive might not be the best choice if the source system's data is semi-structured or unstructured, as they are primarily designed for structured data.

2:  Sqoop requires more manual configuration and setup compared to Spark, which might require more effort.

 Creating a Pipeline with Spark:

Apache Spark is a powerful and versatile framework for distributed data processing that can be used for data ingestion, transformation, and analysis.

Pros:

1: Spark can handle a wide range of data formats, including structured, semi-structured, and unstructured data.

2: It provides a unified data processing engine, which means you can use Spark for both data ingestion and subsequent data processing tasks.

3: Spark is highly scalable and can handle large volumes of data efficiently.

4: It offers a rich ecosystem of libraries and connectors for various data sources.

Cons:

1: Spark might have a steeper learning curve compared to Sqoop and Hive, especially if you're new to Spark.

 2: Setting up a Spark pipeline can be more complex and resource-intensive than a Sqoop/Hive pipeline,  depending on your infrastructure and expertise.

 

Comparison b/w Sqoop and Spark:

1. Data Volume and Complexity:
• Sqoop: Sqoop is typically used for bulk data transfer between Hadoop and relational databases. It is well-suited for ingesting large volumes of structured data from relational databases into Hadoop/Hive.
• Spark: Spark is a more versatile framework that can handle various data sources and formats, including structured, semi-structured, and unstructured data. It is suitable for both batch and real-time data processing. If your data sources are more diverse and complex, Spark may be a better choice.
2. Real-time vs. Batch:
• Sqoop: Sqoop is primarily designed for batch data ingestion. It is not ideal for real-time data ingestion.
• Spark: Spark Streaming and Structured Streaming allow you to process data in near-real-time. If you require real-time or near-real-time data ingestion and processing, Spark might be a better fit.
3. Ecosystem Compatibility:
• Sqoop: Sqoop integrates well with the Hadoop ecosystem, especially Hive. If you are already using Hive and Hadoop extensively, Sqoop can be a seamless choice for data ingestion into Hive tables.
• Spark: Spark is a part of the Hadoop ecosystem and can easily integrate with various Hadoop components, including Hive, HBase, and more. It also offers a broader range of data processing capabilities beyond just ingestion.
4. Data Transformation and Enrichment:
• Sqoop: Sqoop focuses primarily on data transfer and ingestion. If you need to perform data transformations, enrichment, or other data processing tasks during ingestion, you may need to use additional tools or scripts in conjunction with Sqoop.
• Spark: Spark provides powerful data processing capabilities, allowing you to perform transformations and enrichments on the ingested data as part of the same pipeline.
5. Maintenance and Scalability:
• Sqoop: Sqoop is relatively simple to set up and use for basic data ingestion tasks. However, it may require additional tools or scripts for more complex scenarios.
• Spark: Spark offers a more comprehensive solution for data ingestion, processing, and analytics. It can be more scalable and easier to maintain if you have a complex data pipeline.

 

Ultimately, the choice between Sqoop/Hive and Spark for data ingestion depends on your specific use case, the nature of the source data, your existing infrastructure, and your team's expertise. If you need to ingest structured data from relational databases into a Hive-based data warehouse, Sqoop and Hive might be a suitable choice. However, if you have diverse data sources, require real-time or batch processing, or need more flexibility in handling different data formats, Spark is a more versatile option.

 

 

    Magic of “Partition” in “Apache Spark”

What is the “Importance” of “Partition”?

• “Apache Spark” is known for its “Speed”. The “Fast Speed” of “Computing” comes from the “Parallel Processing”.
  

• Partition” is the “Key” for “Parallel Processing”.
• If the “Data”, to work with, is “Partitioned” in a “Proper Way” then the “Query Performance” on that “Data” would be “Improved” as the “Parallel Processing” will be “Triggered” “Effectively”.
• If the “Data”, to work with, is “Not Partitioned” in a “Proper Way” then the “Distributed Framework” of “Apache Spark” is “Not” being used “Effectively”.
• So, “Partition” plays an “Important Role” in the following -
  1. Performance Improvement
  2. Error Handling
  3. Debugging

Why “Partition Strategy” is Needed?

“Partition” is the “Key” for “Parallel Processing” in “Apache Spark”, which makes “Use” of the “Distributed Framework” in “Apache Spark” “Effectively”.

Hence, the “Best Partition Strategy” needs to be “Adopted” to “Achieve” the “Best Performance” in a “Spark Application”.

There are “Two Important Concepts” in “Choosing” the “Correct Partition Strategy” -

1. “Choosing” the “Right Number” of “Partitions”:

• If the “Data” to work with is “Distributed” in the “Right Number” of “Partitions”, “Based” on the “Number of Cores” available in a “Cluster”, then the “Performance” of a “Spark Application” “Boosted”.
• If the “Data” to work with is “Distributed” in the “Wrong Number” of “Partitions”, then the “Performance” of the “Spark Application” is “Decreased”.

2. “Choosing” the “Right Size” of “Partition”:

• If the “Data” to work with is “Distributed” in the “Partitions” of the “Same Size”, i.e, “Evenly Distributed Partitions”, then the “Performance” of a “Spark Application” is “Improved”.
• If the “Data” to work with is “Not Distributed” in the “Partitions” of the “Same Size”, i.e, “Unevenly Distributed Partitions”, then the “Performance” of a “Spark Application” is “Decreased”.

How to “Choose” the “Right Number” of “Partitions”?

The “Right Number” of “Partitions” can be “Calculated” using either of the following way -

1. The “Total Number” of “Cores” available in a “Cluster”

2. “Multiples” of the “Total Number” of “Cores” available in a “Cluster”

Example: Suppose, in a “Cluster”, there are “8 Executors” available, and, “Each” of the “Executors” has “8 Core”. So, over all, there are “64 Cores” available in the “Cluster” in “Each Iteration” of the “Data Processing”.

• Wrong Partition Number”: But, the “Data” to work with is having only “40 Partitions”.
  Now, when an “Action” is “Triggered” on the “Data”, out of the available “64 Cores”, only “40 Cores” will pick “One Partition” of the “Data” to work with “Each”, and, “Start Processing” in the “First Iteration”, but, the “Rest” of the “24 Cores” will be “Idle” in the “First Iteration” itself.
  So, “All” the available “Cores” are “Not Used Effectively” for “Data Processing”.
• Right Partition Number”: So, in this case, the “Right Number” of “Partitions” would be “64”, or, the “Multiples” of “64”, like “128”, or, “192”, and, so on.
  Suppose, if the “Number” of “Partitions” of the “Data” to work with is “128”, then, in the “First Iteration”, “All” of the “64 Cores” will pick up the “64 Partitions” of the “Data” to work with, and, “Start Processing”. Once, the “First Iteration” is “Completed”, then, in the “Second Iteration”, “All” of the “64 Cores” will again pick up the “Next” “64 Partitions” of the “Data” to work with, and, “Start Processing”.
  In this way, “No Core” in the “Cluster” will be “Idle” in “Any Iteration” of the “Data Processing”.

So, “Choosing” the “Right Number” of “Partitions” is “Important” so that “No Core”, available in a “Cluster”, “Sits” “Idle” in “Any Iteration” of the “Data Processing”, and, the “Performance” of a “Spark Application” will be “Increased”.

How the “Unevenly Distributed Partitions” Actually “Decrease” the “Performance” of a “Spark Application”?

• Suppose, in a “Cluster”, there are “8 Executors” available, and, “Each” of the “Executors” has “8 Core”. So, over all, there are “64 Cores” available in the “Cluster” in “Each Iteration” of the “Data Processing”, and, the “Data” to work with is having “64 Partitions”.
• But out of the “64 Partitions” of the “Data” to work with, “32 Partitions” are about “1 GB” in “Size”, and, the “Rest” of the “32 Partitions” are about “100 MB” in “Size”, i.e., the “Partitions” of the “Data” to work with are “Unevenly Distributed”.
• Now, when an “Action” is “Triggered” on the “Data”, “All” of the “64 Cores” will pick up the “64 Partitions” of the “Data” to work with, and, “Start Processing”. But, out of the “64 Cores”, “32 Cores” will “Process” the “Data” of “1 GB” in “Size”, whereas, the “Rest” of the “32 Cores” will “Process” the “Data” of “100 MB” in “Size”.
• So, those “32 Cores”, which “Process” the “Data” of “100 MB” in “Size”, will “Complete” the “Data Processing” “Very Quickly” and “Start Sitting Idle” in the “First Iteration”, whereas, the “Other 32 Cores”, which “Process” the “Data” of “1 GB” in “Size”, will take “More Time” to “Complete” the “Data Processing”.
• This way, the “Unevenly Distributed Partitions” of the “Data” to work can make the available “Cores”, in a “Cluster”, “Sit” “Idle” in “Any Iteration” of the “Data Processing”, and, thereby “Decreasing” the “Performance” of that “Spark Application”.

The “Default Number of Partitions” of “Data” that is “Generated” in the “Spark Environment”

• When the “Data” to work with is “Generated” in the “Spark” Environment, the “Default Number of Partitions” are “Created” from the “Default Value” of the “Spark Configuration”, i.e., “sc.defaultParallelism”.
• The “Default Value” for the “Spark Configuration”, i.e., “sc.defaultParallelism” is “8”. So, by default, when the “Data” to work with is “Generated” in the “Spark” Environment, it is “Distributed” in “8 Partitions”.

# Create the List of the Column Names for the DataFrame
employeeColumns = ["Employee_Id", "First_Name", "Last_Name", "House_No", "Address", "City", "Pincode"]

# Create the List for the Data of the DataFrame
employeeList = [\
                  (1001, "Oindrila", "Chakraborty", "118/H", "Narikel Danga North Road", "Kolkata", 700011),\
                  (1002, "Soumyajyoti", "Bagchi", "38", "Dhakuria East Road", "Kolkata", 700078),\
                  (1003, "Oishi", "Bhattacharyya", "28B", "M.G Road", "Pune", 411009),\
                  (1004, "Sabarni", "Chakraborty", "109A", "Ramkrishna Road", "Kolkata", 700105)\
               ]
              
# Create the DataFrame from the List
employeeColumnNamesDf = spark.createDataFrame(employeeList, schema = employeeColumns)

# Display the "Default Number of Partitions" of the DataFrame
print(sc.defaultParallelism)
print(employeeColumnNamesDf.rdd.getNumPartitions())

Default Number of Partitions of Data Created in Spark Environment

Output -

Since, there is “Not Enough Data” as there are “Only 4 Rows”, out of the “Default 8 Partitions”, “4 Partitions” are “Created” as “Empty”.

employeeColumnNamesDf.rdd.glom().collect()

Display the “Data” from “Each Partitions” of the “DataFrame”

Output -

The “Default Number” and “Default Size” of “Partitions” of “Data” that is “Read” from the “External Storage”

• When the “Data” to work with is “Read” from an “External Storage” to the “Spark Cluster”, the “Number of Partitions” and the “Max Size” of “Each Partition” are “Dependent” on the “Default Value” of the “Spark Configuration”, i.e., “spark.sql.files.maxPartitionBytes”.
• The “Default Value” for the “Spark Configuration”, i.e., “spark.sql.files.maxPartitionBytes” is “128 MB”.
• So, if a “File” of “1 GB” in “Size” is “Read” from an “External Storage”, the “Spark Configuration”, i.e., “spark.sql.files.maxPartitionBytes” will “Start Packing” “128 MB” of “Data” into “Each Partition” at “Maximum”, which makes the “1 GB” of “Data” to be “Distributed” into “8 Partitions” (“1024 MB / 128 MB”), onto the “Spark Cluster”.

df_ReadCustomerFileWithHeader = spark.read\
                                     .option("header", "true")\
                                     .csv("dbfs:/FileStore/tables/retailer/data/customer.csv")

# Display the "Default Value" of the "Spark Configuration", i.e, "spark.sql.files.maxPartitionBytes"
print(spark.conf.get("spark.sql.files.maxPartitionBytes"))

# Display the "Size" of the "CSV File"
print(dbutils.fs.ls("dbfs:/FileStore/tables/retailer/data/customer.csv")) #19900792

# Display the "Default Number of Partitions" of the DataFrame
print(df_ReadCustomerFileWithHeader.rdd.getNumPartitions())

Read a File from External Source

Output -

df_ReadCustomerFileWithHeader.rdd.glom().collect()

Display the “Data” from “Each Partitions” of the “DataFrame”

Output -

Can the “Default Number of Partitions” of “Data” that is “Read” from the “External Storage” be Changed?

Yes. The “Spark Configuration”, i.e., “spark.sql.files.maxPartitionBytes” is “Configurable”. meaning the “Max Size” of “Each Partition” can be “Changed” by “Changing” the “Value” of the “Spark Configuration”, i.e., “spark.sql.files.maxPartitionBytes” depending on the “Different Use Cases” of the “Spark Application”.

Example:

• Some “Developers” may “Develop” a “Spark Application” that can “Handle” “Terabytes” of “Data”. So, the “Max Size” of “Each Partition” can be “Bigger” than the “Default Value” of the “Spark Configuration”, i.e., “spark.sql.files.maxPartitionBytes”.
  In this case, the “New Value” of the “Spark Configuration”, i.e., “spark.sql.files.maxPartitionBytes” can be “Set” as the following -
  spark.conf.set (“spark.sql.files.maxPartitionBytes”, “10000b”).

# Change the "Value" of the "Spark Configuration", i.e, "spark.sql.files.maxPartitionBytes" to "10 KB", i.e., "10000b"
spark.conf.set ("spark.sql.files.maxPartitionBytes", "10000b")

# Display the "Changed Value" of the "Spark Configuration", i.e, "spark.sql.files.maxPartitionBytes"
print(spark.conf.get("spark.sql.files.maxPartitionBytes"))

Change the “Default Value” of the “Spark Configuration”, i.e, “spark.sql.files.maxPartitionBytes”

Output -

df_ReadCustomerFileWithHeader = spark.read\
                                     .option("header", "true")\
                                     .csv("dbfs:/FileStore/tables/retailer/data/customer.csv")

# Display the "Changed Value" of the "Spark Configuration", i.e, "spark.sql.files.maxPartitionBytes"
print(spark.conf.get("spark.sql.files.maxPartitionBytes"))

# Display the "Size" of the "CSV File"
print(dbutils.fs.ls("dbfs:/FileStore/tables/retailer/data/customer.csv")) #19900792

# Display the "Current Number of Partitions" of the DataFrame
print(df_ReadCustomerFileWithHeader.rdd.getNumPartitions())

After Changing the “Default Value” of the “Spark Configuration”, i.e, “spark.sql.files.maxPartitionBytes” to “10KB”, Read a File from External Source

Output -

 

For Which Type of “Files” that is “Read” from the “External Storage”, the “Default Number of Partitions” of “Data” can “Not” be “Applicable”?

• If the “File” that is “Read” from an “External Storage” is “Not” in “Splittable Form”, i.e., if the “File” is “Compressed”, only then the “Data” can “Not” be “Distributed” into the “Number of Partitions” based on the “Default Value” of the “Spark Configuration”, i.e, “spark.sql.files.maxPartitionBytes”.
• In such cases, even if the “Size” of the “File” is “1 TB”, the “Data” would be having just “1 Partition”.
• “Creating” a “Single Partition” with “All” the “Data” of a “File” is “Not Good” for “Performance”, as “Only One Core” would “Process” the “Entire Data” of that “File”, while “All Other Cores” are “Kept Idle”.

Spark Performance Optimization

Things to focus on : 1:Code 2: File Format 3: Configuration 4: Spark optimizations

Big Data File Formats

 Introduction

For data lakes, in the Hadoop ecosystem, HDFS file system is used. However, most cloud providers have replaced it with their own deep storage system such as S3 or GCS. When using deep storage choosing the right file format is crucial.

These file systems or deep storage systems are cheaper than data bases but just provide basic storage and do not provide strong ACID guarantees.

You will need to choose the right storage for your use case based on your needs and budget. For example, you may use a database for ingestion if you budget permit and then once data is transformed, store it in your data lake for OLAP analysis. Or you may store everything in deep storage but a small subset of hot data in a fast storage system such as a relational database.

File Formats

Note that deep storage systems store the data as files and different file formats and compression algorithms provide benefits for certain use cases. How you store the data in your data lake is critical and you need to consider the format, compression and especially how you partition your data.

The most common formats are CSV, JSON, AVRO, Protocol Buffers, Parquet, and ORC.

File Format Options

Some things to consider when choosing the format are:

• The structure of your data: Some formats accept nested data such as JSON, Avro or Parquet and others do not. Even, the ones that do, may not be highly optimized for it. Avro is the most efficient format for nested data, I recommend not to use Parquet nested types because they are very inefficient. Process nested JSON is also very CPU intensive. In general, it is recommended to flat the data when ingesting it.
• Performance: Some formats such as Avro and Parquet perform better than other such JSON. Even between Avro and Parquet for different use cases one will be better than others. For example, since Parquet is a column based format it is great to query your data lake using SQL whereas Avro is better for ETL row level transformation.
• Easy to read: Consider if you need people to read the data or not. JSON or CSV are text formats and are human readable whereas more performant formats such parquet or Avro are binary.
• Compression: Some formats offer higher compression rates than others.
• Schema evolution: Adding or removing fields is far more complicated in a data lake than in a database. Some formats like Avro or Parquet provide some degree of schema evolution which allows you to change the data schema and still query the data. Tools such Delta Lake format provide even better tools to deal with changes in Schemas.
• Compatibility: JSON or CSV are widely adopted and compatible with almost any tool while more performant options have less integration points.

File Formats

• CSV: Good option for compatibility, spreadsheet processing and human readable data. The data must be flat. It is not efficient and cannot handle nested data. There may be issues with the separator which can lead to data quality issues. Use this format for exploratory analysis, POCs or small data sets.
• JSON: Heavily used in APIs. Nested format. It is widely adopted and human readable but it can be difficult to read if there are lots of nested fields. Great for small data sets, landing data or API integration. If possible convert to more efficient format before processing large amounts of data.
• Avro: Great for storing row data, very efficient. It has a schema and supports evolution. Great integration with Kafka. Supports file splitting. Use it for row level operations or in Kafka. Great to write data, slower to read.
• Protocol Buffers: Great for APIs, especially for gRPC. Supports Schema and it is very fast. Use for APIs or machine learning.
• Parquet: Columnar storage. It has schema support. It works very well with Hive and Spark as a way to store columnar data in deep storage that is queried using SQL. Because it stores data in columns, query engines will only read files that have the selected columns and not the entire data set as opposed to Avro. Use it as a reporting layer.
• ORC: Similar to Parquet, it offers better compression. It also provides better schema evolution support as well, but it is less popular.

File Compression

Lastly, you need to also consider how to compress the data considering the trade off between file size and CPU costs. Some compression algorithms are faster but with bigger file size and others slower but with better compression rates. For more details check this article.

Compression options (image by author)

I recommend using snappy for streaming data since it does not require too much CPU power. For batch bzip2 is a great option.

Conclusion

As we can see, CSV and JSON are easy to use, human readable and common formats but lack many of the capabilities of other formats, making it too slow to be used to query the data lake. ORC and Parquet are widely used in the Hadoop ecosystem to query data whereas Avro is also used outside of Hadoop, especially together with Kafka for ingestion, it is very good for row level ETL processing. Row oriented formats have better schema evolution capabilities than column oriented formats making them a great option for data ingestion.