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”.