Ahmed Ayed

𝐃𝐞𝐚𝐫 𝐃𝐚𝐭𝐚 𝐄𝐧𝐠𝐢𝐧𝐞𝐞𝐫𝐬 🔥 Here I come with another PySpark Transformation Challenge 📣 𝐈𝐧𝐟𝐨: You have a DataFrame with two columns: 𝐢𝐝 and 𝐩𝐥𝐚𝐧_𝐝𝐚𝐭𝐞. 𝐘𝐨𝐮𝐫 𝐭𝐚𝐬𝐤 𝐢𝐬 𝐭𝐨 𝐜𝐫𝐞𝐚𝐭𝐞 𝐭𝐡𝐫𝐞𝐞 𝐧𝐞𝐰 𝐜𝐨𝐥𝐮𝐦𝐧𝐬: 1. 𝐈𝐒_𝐋𝐀𝐒𝐓_𝐒𝐈𝐗_𝐌𝐎𝐍𝐓𝐇𝐒: For each id, check whether the date is within the last 6 months compared to the previous row for that id. The first row for each id should always be "YES", and subsequent rows should check if the previous date is less than 6 months before. If it is within 6 months, mark it as "YES", otherwise, mark it as "NO". 2. 𝐈𝐒_𝐂𝐔𝐑𝐑𝐄𝐍𝐓_𝐅𝐈𝐍𝐀𝐍𝐂𝐈𝐀𝐋_𝐘𝐄𝐀𝐑: For each id, check whether the date belongs to the current financial year (April 1, 2024 - March 31, 2025). 3. 𝐈𝐒_𝐂𝐔𝐑𝐑𝐄𝐍𝐓_𝐘𝐄𝐀𝐑: For each id, check whether the date belongs to the current calendar year (January 1, 2024 - December 31, 2024). 𝐂𝐫𝐞𝐚𝐭𝐞 𝐒𝐚𝐦𝐩𝐥𝐞 𝐃𝐚𝐭𝐚𝐟𝐫𝐚𝐦𝐞 (Input Data): 𝘥𝘢𝘵𝘢 = [   (1, "2024-08-15"),   (1, "2024-03-10"),   (1, "2023-05-05"),   (2, "2025-02-24"),   (2, "2024-07-15"),   (2, "2024-03-15") ] 𝘤𝘰𝘭𝘶𝘮𝘯𝘴 = ["𝘪𝘥", "𝘱𝘭𝘢𝘯_𝘥𝘢𝘵𝘦"] 𝘥𝘧 = 𝘴𝘱𝘢𝘳𝘬.𝘤𝘳𝘦𝘢𝘵𝘦𝘋𝘢𝘵𝘢𝘍𝘳𝘢𝘮𝘦(𝘥𝘢𝘵𝘢, 𝘤𝘰𝘭𝘶𝘮𝘯𝘴) 𝐏𝐲𝐒𝐩𝐚𝐫𝐤 𝐓𝐫𝐚𝐧𝐬𝐟𝐨𝐫𝐦𝐚𝐭𝐢𝐨𝐧 𝐂𝐨𝐝𝐞? (Expected Data) Let me know the answer in the comments 😀 #pyspark #dataengineering

81

6 commentaires

🚀 PySpark Data Engineer Interview Questions Set (Intermediate Level)🔥 🔍 𝐈𝐧𝐭𝐫𝐨𝐝𝐮𝐜𝐭𝐢𝐨𝐧: 1️⃣ Can you provide an overview of your experience with PySpark and big data processing?   2️⃣ What motivated you to specialize in PySpark, and how have you applied it in your previous roles?  ⚙️ 𝐏𝐲𝐒𝐩𝐚𝐫𝐤 𝐁𝐚𝐬𝐢𝐜𝐬:  3️⃣ Explain the basic architecture of PySpark.   4️⃣ How does PySpark relate to Apache Spark?  💻 𝐃𝐚𝐭𝐚𝐅𝐫𝐚𝐦𝐞 𝐎𝐩𝐞𝐫𝐚𝐭𝐢𝐨𝐧𝐬:  5️⃣ Difference between DataFrame and RDD?   6️⃣ Explain transformations and actions in PySpark DataFrames.  🚀 𝐎𝐩𝐭𝐢𝐦𝐢𝐳𝐢𝐧𝐠 𝐏𝐲𝐒𝐩𝐚𝐫𝐤 𝐉𝐨𝐛𝐬: 8️⃣ How do you optimize PySpark job performance?   9️⃣ Techniques for handling skewed data?  📦 𝐃𝐚𝐭𝐚 𝐒𝐞𝐫𝐢𝐚𝐥𝐢𝐳𝐚𝐭𝐢𝐨𝐧 𝐚𝐧𝐝 𝐂𝐨𝐦𝐩𝐫𝐞𝐬𝐬𝐢𝐨𝐧:  🔟 Explain data serialization in PySpark.   1️⃣1️⃣ Significance of choosing the right compression codec?  🧠 𝐏𝐲𝐒𝐩𝐚𝐫𝐤 𝐌𝐚𝐜𝐡𝐢𝐧𝐞 𝐋𝐞𝐚𝐫𝐧𝐢𝐧𝐠: 1️⃣8️⃣ Discuss your experience with PySpark’s MLlib.   1️⃣9️⃣ Examples of machine learning algorithms you’ve implemented? 🌐 𝐈𝐧𝐭𝐞𝐠𝐫𝐚𝐭𝐢𝐨𝐧 𝐰𝐢𝐭𝐡 𝐎𝐭𝐡𝐞𝐫 𝐓𝐞𝐜𝐡𝐧𝐨𝐥𝐨𝐠𝐢𝐞𝐬:   2️⃣2️⃣ Have you integrated PySpark with other big data technologies?  🔧 𝐑𝐞𝐚𝐥-𝐰𝐨𝐫𝐥𝐝 𝐏𝐫𝐨𝐣𝐞𝐜𝐭 𝐒𝐜𝐞𝐧𝐚𝐫𝐢𝐨: 2️⃣5️⃣ Describe a challenging PySpark project you’ve worked on and how you overcame the challenges.  [Happy Learning 📚] 👉Follow Aishwarya Pani(She/Her) for more data engineering-related materials and information. Save and reshare ✅ 𝑹𝒆𝒑𝒐𝒔𝒕 𝒊𝒇 𝒚𝒐𝒖 𝒇𝒊𝒏𝒅 𝒊𝒕 𝒖𝒔𝒆𝒇𝒖𝒍 #PySpark #BigData #DataEngineering #MachineLearning #Spark #DataFrames #ClusterManagement #JobOptimization #TechInterview #DataScience

20

Here's a secret Snowflake tip 🤫: with the ability to host Notebooks on the new container runtime, we can now install arbitrary packages using !pip. 🔥 Yes, we can also install packages not available on the Anaconda Snowflake repo (with all the risks that come with it 😅). For instance, now you can install the Cinemagoer package and search the IMDb for all the movies containing the word "matrix" in the title 👇

89

Explain the difference between repartition() and coalesce() in PySpark: My answer: Challenge accepted, let’s go! ➤ 𝗿𝗲𝗽𝗮𝗿𝘁𝗶𝘁𝗶𝗼𝗻() 𝘁𝗿𝗮𝗻𝘀𝗳𝗼𝗿𝗺𝗮𝘁𝗶𝗼𝗻: The repartition() transformation is used to increase or decrease the number of partitions in a DataFrame or RDD. It performs a full shuffle of data across all nodes in the cluster, which can be costly but ensures even distribution of data. # Example DataFrame df = spark.createDataFrame([("Alice", 29), ("Bob", 34), ("Cathy", 28), ("David", 45)]) # Original number of partitions print(df.rdd.getNumPartitions()) # Output might be 4 (default) # Increase the number of partitions to 6 using repartition() repartitioned_df = df.repartition(6) # Check the number of partitions after repartitioning print(repartitioned_df.rdd.getNumPartitions()) # Output: 6 ➤ 𝗰𝗼𝗮𝗹𝗲𝘀𝗰𝗲() 𝘁𝗿𝗮𝗻𝘀𝗳𝗼𝗿𝗺𝗮𝘁𝗶𝗼𝗻: The coalesce() transformation is used to reduce the number of partitions in a DataFrame or RDD. It avoids a full shuffle by moving data only between fewer partitions, making it more efficient than repartition() when reducing partitions. # Example DataFrame df = spark.createDataFrame([("Alice", 29), ("Bob", 34), ("Cathy", 28), ("David", 45)]) # Original number of partitions print(df.rdd.getNumPartitions()) # Output might be 4 (default) # Reduce the number of partitions to 2 using coalesce() coalesced_df = df.coalesce(2) # Check the number of partitions after coalescing print(coalesced_df.rdd.getNumPartitions()) # Output: 2 𝗞𝗲𝘆 𝗗𝗶𝗳𝗳𝗲𝗿𝗲𝗻𝗰𝗲: - repartition() involves a full shuffle and is used to both increase and decrease partitions. - coalesce() minimizes data movement by avoiding a full shuffle and is used primarily to decrease the number of partitions.

28

1 commentaire

𝐖𝐡𝐲 𝐰𝐨𝐮𝐥𝐝 𝐨𝐧𝐞 𝐧𝐞𝐞𝐝 𝐚 𝐝𝐚𝐭𝐚𝐰𝐚𝐫𝐞𝐡𝐨𝐮𝐬𝐞 𝐥𝐢𝐤𝐞 #redshift when you can directly run #sql on a DataLake on #S3 ? 𝐁𝐮𝐭 𝐡𝐨𝐰 𝐜𝐚𝐧 𝐨𝐧𝐞 𝐪𝐮𝐞𝐫𝐲 𝐝𝐚𝐭𝐚 𝐥𝐚𝐤𝐞 𝐮𝐬𝐢𝐧𝐠 #sql ? Of course, using #athena via #glue catalog 𝐖𝐡𝐚𝐭'𝐬 𝐠𝐥𝐮𝐞 𝐜𝐚𝐭𝐚𝐥𝐨𝐠? Heard of Big Data thing - Hive Metastore? Glue Catalog is just like that -- just holds the metadata about the tables. 𝐍𝐨𝐰 𝐭𝐨 𝐚𝐧𝐬𝐰𝐞𝐫 𝐭𝐡𝐞 𝐨𝐫𝐢𝐠𝐢𝐧𝐚𝐥 𝐪𝐮𝐞𝐬𝐭𝐢𝐨𝐧 - Redshift is a Massively parallel processing Database that can scale horizontally like how you can have nodes in a big data cluster. And so, it's very very powerful and can get you run your analytics on petabytes of data at a much faster pace!! These queries are generally used in the "time critical" reports created on aws #quicksight or any other tool like Tableau , PowerBI. 𝐍𝐨𝐭𝐞 𝐭𝐡𝐞 𝐰𝐨𝐫𝐝 - 𝐭𝐢𝐦𝐞 𝐜𝐫𝐢𝐭𝐢𝐜𝐚𝐥 - 𝐓𝐡𝐚𝐭'𝐬 𝐰𝐡𝐲 𝐲𝐨𝐮 𝐧𝐞𝐞𝐝 #redshift! #aws #dataengineering ✅Version 3 of 𝐬𝐜𝐞𝐧𝐚𝐫𝐢𝐨 𝐛𝐚𝐬𝐞𝐝 𝐬𝐭𝐫𝐮𝐜𝐭𝐮𝐫𝐞𝐝 𝐡𝐚𝐧𝐝𝐬-𝐨𝐧 𝐑𝐞𝐚𝐥-world AWS Data Engineering ( RADE™) program started on July 28th; 2 𝐒𝐎𝐋𝐈𝐃 𝐇𝐀𝐍𝐃𝐒 𝐎𝐍 𝐒𝐄𝐒𝐒𝐈𝐎𝐍 𝐀𝐑𝐄 𝐃𝐎𝐍𝐄! ✅𝐈𝐟 𝐲𝐨𝐮 𝐚𝐫𝐞 𝐚𝐜𝐭𝐢𝐨𝐧-𝐨𝐫𝐢𝐞𝐧𝐭𝐞𝐝 and want to get in, DM me NOW. 𝐉𝐨𝐢𝐧 𝐚𝐥𝐨𝐧𝐠 𝐰𝐢𝐭𝐡 80 𝐚𝐦𝐚𝐳𝐢𝐧𝐠 𝐩𝐞𝐨𝐩𝐥𝐞 𝐰𝐡𝐨 𝐡𝐚𝐯𝐞 𝐞𝐧𝐫𝐨𝐥𝐥𝐞𝐝.

57

4 commentaires

🚀 New to Spark? These concepts will help you build a solid foundation! 💡 🔗 Pyspark Inner Workings : https://lnkd.in/dkeuuGw3 🔗 Partitioning in Spark : https://lnkd.in/dG7wmBN4 🔗 Custom Partitioning : https://lnkd.in/d6xH8MiQ 🔗 Shuffling in Spark : https://lnkd.in/dKPvfpuD 🔗 Dealing with Null Values : https://lnkd.in/d9B8XcWY 🔗 Understanding Garbage Collection : https://lnkd.in/degCsfsg These topics will help you understand Spark's inner workings, giving you the knowledge to identify potential issues and optimize your jobs more effectively. Is it just me, or do all production issues occur on Fridays? 😂 #bigdata #data #spark

11

𝐀𝐜𝐞 𝐘𝐨𝐮𝐫 𝐏𝐲𝐒𝐩𝐚𝐫𝐤 𝐈𝐧𝐭𝐞𝐫𝐯𝐢𝐞𝐰𝐬 𝐰𝐢𝐭𝐡 𝐓𝐡𝐞𝐬𝐞 20 𝐑𝐞𝐚𝐥-𝐓𝐢𝐦𝐞, 𝐒𝐜𝐞𝐧𝐚𝐫𝐢𝐨-𝐁𝐚𝐬𝐞𝐝 𝐐𝐮𝐞𝐬𝐭𝐢𝐨𝐧𝐬! 🚀 Are you preparing for PySpark interviews at top companies? Get ready with these hands-on questions designed to evaluate your real-world experience! 🔥 𝐋𝐞𝐭'𝐬 𝐛𝐫𝐞𝐚𝐤 𝐢𝐭 𝐝𝐨𝐰𝐧 𝐢𝐧𝐭𝐨 4 𝐤𝐞𝐲 𝐚𝐫𝐞𝐚𝐬: 1. 𝐃𝐚𝐭𝐚 𝐏𝐫𝐨𝐜𝐞𝐬𝐬𝐢𝐧𝐠 𝐚𝐧𝐝 𝐓𝐫𝐚𝐧𝐬𝐟𝐨𝐫𝐦𝐚𝐭𝐢𝐨𝐧 1️⃣ How would you handle large datasets in PySpark and optimize for performance? 2️⃣ How do you efficiently join two massive datasets (e.g., 100GB each) in PySpark? 3️⃣ Faced with millions of records, how do you identify and remove duplicates using PySpark? 4️⃣ Given a DataFrame with nested JSON, how would you flatten it in PySpark? 5️⃣ How do you handle missing or null values in a DataFrame? What strategies work best? 2. 𝐏𝐞𝐫𝐟𝐨𝐫𝐦𝐚𝐧𝐜𝐞 𝐓𝐮𝐧𝐢𝐧𝐠 𝐚𝐧𝐝 𝐎𝐩𝐭𝐢𝐦𝐢𝐳𝐚𝐭𝐢𝐨𝐧 6️⃣ How do you debug and optimize PySpark jobs that are taking too long to finish? 7️⃣ What is a shuffle operation in PySpark, and how do you minimize its performance impact? 8️⃣ Have you dealt with data skew in PySpark? What steps did you take to resolve it? 9️⃣ How do you optimize PySpark jobs in a YARN cluster environment? 🔟 What’s the difference between repartition() and coalesce()? When should you use each? 3. 𝐃𝐚𝐭𝐚 𝐏𝐢𝐩𝐞𝐥𝐢𝐧𝐞 𝐃𝐞𝐯𝐞𝐥𝐨𝐩𝐦𝐞𝐧𝐭 1️⃣1️⃣ How would you build an ETL pipeline in PySpark to process streaming data? 1️⃣2️⃣ How do you ensure data consistency and fault tolerance in PySpark jobs? 1️⃣3️⃣ You need to aggregate data from multiple sources and save it as a partitioned Parquet file. How would you achieve this? 1️⃣4️⃣ How do you orchestrate and manage a complex PySpark job with multiple stages? 1️⃣5️⃣ How do you handle schema evolution in PySpark while reading and writing data? 4. 𝐃𝐞𝐛𝐮𝐠𝐠𝐢𝐧𝐠 𝐚𝐧𝐝 𝐄𝐫𝐫𝐨𝐫 𝐇𝐚𝐧𝐝𝐥𝐢𝐧𝐠 1️⃣6️⃣ Have you encountered out-of-memory errors in PySpark? How did you solve them? 1️⃣7️⃣ What steps would you take if a PySpark job fails midway? How do you recover? 1️⃣8️⃣ A Spark task keeps failing due to data corruption in one partition. How do you handle it? 1️⃣9️⃣ Have you used custom UDFs in PySpark? What challenges did you face, and how did you overcome them? 2️⃣0️⃣ Have you ever debugged a PySpark job producing incorrect results? What was your approach? ✨ Follow for more interview insights and tips to ace your next PySpark challenge! ✨ #SQL #BigData #DataEngineering #Hadoop #HDFS #Sqoop #Hive #Spark #Snowflake #AWS #Azure #GCP #Interview #PySpark

75

Creating Apache Hudi tables on S3 Data Lake storage. Want to get started with creating Hudi tables in a cloud data lake like S3 (or a S3 compatible one like MinIO) ? Here are all the configurations you will need to connect: ✅ Spark SQL - compute ✅ Hudi - Open Table format ✅ S3 - Data Lake Code link in comments. #dataengineering #softwareengineering

105

3 commentaires

🚀 DE Interview Experience 📢 Apache Spark ⌛ 1-Min-Read How to Use Semi Join for Inserting Delta Records and Data Testing in Spark 👉🏽 🧔🏽‍♂️Interviewer: Can you explain how Semi Join can be used for inserting only delta records into Hive tables in Apache Spark and also to do data testing when comparing two datasets for data analysis? 👨‍🦰Interviewee: A Semi Join in Spark is used to filter the rows in one dataset based on whether they have matching keys in another dataset. This can be very helpful for identifying delta records and for comparing two datasets. 🧔🏽‍♂️Interviewer: Can you walk me through how you would implement this for inserting delta records into Hive tables? 👨‍🦰Interviewee: Yes, to insert only delta records into Hive tables using Semi Join, you first need to identify the delta records. Here's a step-by-step process: Load the Source and Target Data: Load the current dataset from the source and the existing dataset from the target Hive table. Perform the Semi Join: Use a left anti join or left semi join to find records that are present in the source dataset but not in the target dataset. This will give us the delta records. val deltaRecords = sourceData.join(targetData, Seq("keyColumn"), "leftanti") Insert Delta Records into Hive Table: Insert the identified delta records into the target Hive table. deltaRecords.write.mode("append").saveAsTable("targetHiveTable") 🧔🏽‍♂️Interviewer: That's a clear explanation. Now, how would you use Semi Join for data testing when comparing two datasets? 👨‍🦰Interviewee: For data testing, Semi Join can be used to compare two datasets to find discrepancies. Here's the step-by-step process: Load Both Datasets: Load the datasets that you want to compare. Identify Missing Records: Use a left anti join to find records that are present in one dataset but missing in the other. val missingInDataset1 = dataset2.join(dataset1, Seq("keyColumn"), "leftanti") val missingInDataset2 = dataset1.join(dataset2, Seq("keyColumn"), "leftanti") Analyze Differences: By analyzing the results of these joins, you can understand which records are missing or different between the two datasets, which helps in data validation and testing. 🧔🏽‍♂️Interviewer: Excellent example! What are the main benefits of using Semi Join for these tasks? 👨‍🦰Interviewee: The main benefits are: Efficiency: It reduces the amount of data processed, as it only considers relevant records. Accuracy: Helps ensure that only the required delta records are inserted or updated, maintaining data integrity. Scalability: Semi Join operations are scalable and can handle large datasets effectively, making them suitable for big data environments. #DataEngineering #ApacheSpark #Hive #DataTesting #DeltaRecords #SemiJoin #DataAnalysis #BigData #ETL #TechInnovation #OneStepAnalytics

13

In case you missed the release of the #MySQL Connector/Python 9.0.0, it deprecated several cursors, updated X DevAPI, and removed Debian, Solaris, and most RPM (other than Enterprise Linux). Now, you install Connector/Python with the pip utility. #python #softwareengineering #programming #devops #cloud #systemengineering #pip https://lnkd.in/gnxaGVnr

2

Scaling Data in Big Data Technologies : In the dynamic world of big data, scaling your data infrastructure is essential for maintaining performance and reliability. Here are some key strategies to help you scale effectively: #Horizontal Scaling: - Add More Nodes: Increase the number of nodes in your cluster to distribute the load. - Distributed Storage: Use systems like Hadoop HDFS or Amazon S3 for seamless scaling. #Partitioning and Sharding: - Data Partitioning: Split large datasets into manageable chunks. - Sharding: Distribute data across multiple databases for load balancing. #Optimization and Tuning: - Query Optimization: Use indexing and caching to speed up data retrieval. - Resource Management: Fine-tune CPU and memory allocation with tools like Kubernetes. #Data Pipeline Scaling: -Stream Processing: Utilize frameworks like Apache Kafka Streams for real-time data. - Batch Processing: Employ Apache Spark for efficient batch processing. #Cloud Solutions: - Scalable Cloud Services: Leverage AWS, GCP, or Azure for scalable storage and compute. - Serverless Architectures: Use AWS Lambda or Google Cloud Functions for automatic scaling. #Monitoring and Management - Continuous Monitoring: Implement tools like Prometheus to track performance. - Automated Management: Use Ansible or Terraform for infrastructure automation. #Data Governance and Security: - Governance Policies: Ensure data quality and compliance. - Security Measures: Implement encryption and access controls to protect data. By following these best practices, you can build a scalable, robust big data environment to meet your growing data needs. #BigData #DataEngineering

9

🚀 DE Interview📢 Apache Spark ⌛ 1-Min-Read How to Use Semi Join for Inserting Delta Records and Testing in Spark 👉🏽 🧔🏽‍♂️Interviewer: Can you explain how Semi Join can be used for inserting only delta records into Hive tables in Apache Spark and also to do data testing when comparing two datasets for data analysis? 👨‍🦰Interviewee: A Semi Join in Spark is used to filter the rows in one dataset based on whether they have matching keys in another dataset. This can be very helpful for identifying delta records and for comparing two datasets. 🧔🏽‍♂️Interviewer: Can you walk me through how you would implement this for inserting delta records into Hive tables? 👨‍🦰Interviewee: Yes, to insert only delta records into Hive tables using Semi Join, you first need to identify the delta records. Here's a step-by-step process: Load the Source and Target Data: Load the current dataset from the source and the existing dataset from the target Hive table. Perform the Semi Join: Use a left anti-join or left semi-join to find records that are present in the source dataset but not in the target dataset. This will give us the delta records. val deltaRecords = sourceData.join(targetData, Seq("keyColumn"), "leftanti") deltaRecords.write.mode("append").saveAsTable("targetHiveTable") Insert Delta Records into Hive Table: Insert the identified delta records into the target Hive table. 🧔🏽‍♂️Interviewer: That's a clear explanation. Now, how would you use Semi Join for data testing when comparing two datasets? 👨‍🦰Interviewee: For data testing, Semi Join can be used to compare two datasets to find discrepancies. Here's the step-by-step process: Load Both Datasets: Load the datasets that you want to compare. Identify Missing Records: Use a left anti join to find records that are present in one dataset but missing in the other. val missingInDataset1 = dataset2.join(dataset1, Seq("keyColumn"), "leftanti") val missingInDataset2 = dataset1.join(dataset2, Seq("keyColumn"), "leftanti") Analyze Differences: By analyzing the results of these joins, you can understand which records are missing or different between the two datasets, which helps in data validation and testing. 📚Summary: 1️⃣ Efficiency: It reduces the amount of data processed, as it only considers relevant records. 2️⃣ Accuracy: Helps ensure that only the required delta records are inserted or updated, maintaining data integrity. 3️⃣ Scalability: Semi Join operations are scalable and can handle large datasets effectively, making them suitable for big data environments. #DataEngineering #ApacheSpark #Hive #DataTesting #DeltaRecords #SemiJoin #DataAnalysis #BigData #ETL #TechInnovation #OneStepAnalytics

23

🚀 DE Interview📢 Apache Spark ⌛ 1-Min-Read How to Use Semi Join for Inserting Delta Records and Testing in Spark 👉🏽 🧔🏽‍♂️Interviewer: Can you explain how Semi Join can be used for inserting only delta records into Hive tables in Apache Spark and also to do data testing when comparing two datasets for data analysis? 👨‍🦰Interviewee: A Semi Join in Spark is used to filter the rows in one dataset based on whether they have matching keys in another dataset. This can be very helpful for identifying delta records and for comparing two datasets. 🧔🏽‍♂️Interviewer: Can you walk me through how you would implement this for inserting delta records into Hive tables? 👨‍🦰Interviewee: Yes, to insert only delta records into Hive tables using Semi Join, you first need to identify the delta records. Here's a step-by-step process: Load the Source and Target Data: Load the current dataset from the source and the existing dataset from the target Hive table. Perform the Semi Join: Use a left anti-join or left semi-join to find records that are present in the source dataset but not in the target dataset. This will give us the delta records. val deltaRecords = sourceData.join(targetData, Seq("keyColumn"), "leftanti") deltaRecords.write.mode("append").saveAsTable("targetHiveTable") Insert Delta Records into Hive Table: Insert the identified delta records into the target Hive table. 🧔🏽‍♂️Interviewer: That's a clear explanation. Now, how would you use Semi Join for data testing when comparing two datasets? 👨‍🦰Interviewee: For data testing, Semi Join can be used to compare two datasets to find discrepancies. Here's the step-by-step process: Load Both Datasets: Load the datasets that you want to compare. Identify Missing Records: Use a left anti join to find records that are present in one dataset but missing in the other. val missingInDataset1 = dataset2.join(dataset1, Seq("keyColumn"), "leftanti") val missingInDataset2 = dataset1.join(dataset2, Seq("keyColumn"), "leftanti") Analyze Differences: By analyzing the results of these joins, you can understand which records are missing or different between the two datasets, which helps in data validation and testing. 📚Summary: 1️⃣ Efficiency: It reduces the amount of data processed, as it only considers relevant records. 2️⃣ Accuracy: Helps ensure that only the required delta records are inserted or updated, maintaining data integrity. 3️⃣ Scalability: Semi Join operations are scalable and can handle large datasets effectively, making them suitable for big data environments. #DataEngineering #ApacheSpark #Hive #DataTesting #DeltaRecords #SemiJoin #DataAnalysis #BigData #ETL #TechInnovation #OneStepAnalytics

17

Datawarehouse Vs Datalake!! Interviewer: let's discuss the differences between data warehouses and data lakes. Are you familiar with these concepts?" Candidate: "Yes, I've worked with both data warehouses and data lakes. They serve as repositories for storing and analyzing large volumes of data, but they have different architectures and use cases." Interviewer: "That's correct! Let's start with data warehouses. How would you define a data warehouse, and what are its key characteristics?" Candidate: "A data warehouse is a centralized repository that stores structured and processed data from various sources. It's optimized for querying and analysis, typically using a schema-on-write approach, where data is structured and organized before being loaded into the warehouse. Data warehouses are designed for supporting business intelligence (BI) and analytics applications, providing fast and reliable access to historical data." Interviewer: " How do data lakes differ from data warehouses, and what are their primary characteristics?" Candidate: "Unlike data warehouses, data lakes store raw, unstructured, or semi-structured data in its native format. They use a schema-on-read approach, where data is ingested without prior structuring, allowing for flexible exploration and analysis. Data lakes are designed to store vast amounts of data at a low cost and support a wide range of data processing and analytics use cases, including data exploration, machine learning, and advanced analytics." Interviewer: "Spot on! Now, let's discuss some key differences between data warehouses and data lakes. What are some factors to consider when deciding between these two architectures?" Candidate: "One significant difference is their approach to data organization and schema management. Data warehouses enforce structure and schema upfront, making them suitable for well-defined, structured data. On the other hand, data lakes offer flexibility and agility, enabling organizations to store diverse data types and schemaless data. Interviewer: "Excellent points! Both data warehouses and data lakes have their strengths and use cases. Now, let's consider a hypothetical scenario: Suppose a company wants to implement a data platform to support both real-time analytics and ad-hoc exploration of raw data. Which approach—data warehouse or data lake—would you recommend, and why?" Candidate: "In this scenario, I would recommend leveraging a data lake architecture. A data lake would provide the flexibility and scalability needed to handle real-time streaming data and accommodate ad-hoc exploration of raw data. With a data lake, the company can ingest data from various sources without worrying about upfront schema definition, enabling faster data ingestion and exploration. Additionally, the company can leverage technologies like Apache Spark or Apache Flink for real-time analytics and processing of streaming data." #dwh #datalake

78

3 commentaires

𝐃𝐞𝐚𝐫 𝐃𝐚𝐭𝐚 𝐄𝐧𝐠𝐢𝐧𝐞𝐞𝐫𝐬 🔥 Happy Monday!! Here I come with another PySpark Transformation Challenge 📣 𝐈𝐧𝐟𝐨: You have a DataFrame with two columns: 𝒆𝒎𝒑𝒍𝒐𝒚𝒆𝒆_ 𝒊𝒅 and 𝒄𝒐𝒏𝒕𝒂𝒄𝒕_𝒅𝒆𝒕𝒂𝒊𝒍𝒔. The 𝒄𝒐𝒏𝒕𝒂𝒄𝒕_𝒅𝒆𝒕𝒂𝒊𝒍𝒔 column contains unstructured text with potential phone numbers (Let's assume 10 digits without country code) and email addresses. 𝐘𝐨𝐮𝐫 𝐭𝐚𝐬𝐤 𝐢𝐬 𝐭𝐨: 1. Extract the phone numbers (10 digits) and email addresses from the contact_details column. 2. Create two new columns, phone_number and email_id, based on the extracted values. 3. If no phone numbers or email addresses are found, the corresponding values should be null. 𝐂𝐫𝐞𝐚𝐭𝐞 𝐒𝐚𝐦𝐩𝐥𝐞 𝐃𝐚𝐭𝐚𝐟𝐫𝐚𝐦𝐞 (Input Data): 𝘥𝘢𝘵𝘢 = [ ("𝘌001", "𝘑𝘰𝘩𝘯 𝘸𝘰𝘳𝘬𝘴 𝘢𝘵 𝘈𝘉𝘊 𝘊𝘰𝘳𝘱. 𝘊𝘰𝘯𝘵𝘢𝘤𝘵: 9876543210"), ("𝘌002", "𝘈𝘯𝘯𝘢'𝘴 𝘦𝘮𝘢𝘪𝘭 𝘪𝘴 𝘢𝘯𝘯𝘢.𝘴𝘮𝘪𝘵𝘩@𝘨𝘮𝘢𝘪𝘭.𝘤𝘰𝘮. 𝘏𝘦𝘳 𝘱𝘩𝘰𝘯𝘦 𝘪𝘴 9123456789"), ("𝘌003", "𝘕𝘰 𝘤𝘰𝘯𝘵𝘢𝘤𝘵 𝘪𝘯𝘧𝘰𝘳𝘮𝘢𝘵𝘪𝘰𝘯 𝘢𝘷𝘢𝘪𝘭𝘢𝘣𝘭𝘦."), ("𝘌004", "𝘙𝘦𝘢𝘤𝘩 𝘮𝘦 𝘢𝘵 9234567890 𝘰𝘳 𝘷𝘪𝘢 𝘮𝘢𝘪𝘭 𝘢𝘭𝘪𝘤𝘦.𝘫𝘰𝘩𝘯𝘴𝘰𝘯@𝘹𝘺𝘻.𝘤𝘰.𝘶𝘬") ] 𝘥𝘧 = 𝘴𝘱𝘢𝘳𝘬.𝘤𝘳𝘦𝘢𝘵𝘦𝘋𝘢𝘵𝘢𝘍𝘳𝘢𝘮𝘦(𝘥𝘢𝘵𝘢, ["𝘦𝘮𝘱𝘭𝘰𝘺𝘦𝘦_𝘪𝘥", "𝘤𝘰𝘯𝘵𝘢𝘤𝘵_𝘥𝘦𝘵𝘢𝘪𝘭𝘴"]) 𝐏𝐲𝐒𝐩𝐚𝐫𝐤 𝐓𝐫𝐚𝐧𝐬𝐟𝐨𝐫𝐦𝐚𝐭𝐢𝐨𝐧 𝐂𝐨𝐝𝐞? (Expected Data) Let me know the answer in the comments 😀 #pyspark #dataengineering

101

14 commentaires

Databricks Data Engineering Interview Question Series 𝑳𝒆𝒕'𝒔 𝑻𝒂𝒍𝒌 𝒂𝒃𝒐𝒖𝒕 𝑺𝒑𝒂𝒓𝒌 𝒎𝒆𝒎𝒐𝒓𝒚 𝒂𝒍𝒍𝒐𝒄𝒂𝒕𝒊𝒐𝒏    Ever wondered how does spark manages its memory allocation? when you specify 8GB memory (for example), what part of it is used for execution and what part is used for caching? 𝐒𝐩𝐚𝐫𝐤 𝐌𝐞𝐦𝐨𝐫𝐲 𝐀𝐥𝐥𝐨𝐜𝐚𝐭𝐢𝐨𝐧 On a high level, spark has 2 kinds of memories:  • Driver memory  • Executor memory 𝐃𝐫𝐢𝐯𝐞𝐫 𝐌𝐞𝐦𝐨𝐫𝐲 𝐀𝐥𝐥𝐨𝐜𝐚𝐭𝐢𝐨𝐧 The driver memory itself has 2 kinds of memories again:  • 𝑯𝒆𝒂𝒑 𝒎𝒆𝒎𝒐𝒓𝒚: Here all the JVM processes runs (and that’ll be your spark code).  • 𝑶𝒗𝒆𝒓𝒉𝒆𝒂𝒅 𝒎𝒆𝒎𝒐𝒓𝒚: Here all the non JVM processes runs. This memory is used for shuffles, exchanges and network buffer. 𝐄𝐱𝐞𝐜𝐮𝐭𝐨𝐫 𝐌𝐞𝐦𝐨𝐫𝐲 𝐀𝐥𝐥𝐨𝐜𝐚𝐭𝐢𝐨𝐧 Coming to executor memory, we have 4 kinds of memories:  • Heap memory  • Overhead memory  • Offheap memory  • pyspark memory Moving ahead, let's talk more about what goes inside Heap memory of an executor:  • 𝑹𝒆𝒔𝒆𝒓𝒗𝒆𝒅 𝒎𝒆𝒎𝒐𝒓𝒚: 300MB and is fixed for spark engine.  • 𝑺𝒑𝒂𝒓𝒌 𝒎𝒆𝒎𝒐𝒓𝒚: 60% of (8GB-300MB). This memory is used for dataframe operations and caching. (This is where the magic happens ✨)  • 𝑼𝒔𝒆𝒓 𝒎𝒆𝒎𝒐𝒓𝒚: 40% of (8GB-300MB). This memory is used for user defined functions, spark internal metadata, RDD conversion operations and RDD lineage and dependencies. The spark memory is further bifurcated in 2 parts:  • 𝑺𝒕𝒐𝒓𝒂𝒈𝒆 𝒎𝒆𝒎𝒐𝒓𝒚 𝒑𝒐𝒐𝒍: cache memory for dataframes  • 𝑬𝒙𝒆𝒄𝒖𝒕𝒐𝒓 𝒎𝒆𝒎𝒐𝒓𝒚 𝒑𝒐𝒐𝒍: This memory is used as buffer memory for dataframe operations. ♻ Feel free to re-share this post if you find it helpful. Follow ABHAY SINGH 😀 𝑪𝒐𝒏𝒏𝒆𝒄𝒕 𝒇𝒐𝒓 1 𝒕𝒐 1 𝒈𝒖𝒊𝒅𝒂𝒏𝒄𝒆 : https://lnkd.in/g7Dewhyq

159

4 commentaires

𝐌𝐢𝐜𝐫𝐨𝐬𝐨𝐟𝐭 𝐅𝐀𝐁𝐑𝐈𝐂 𝐮𝐬𝐞𝐫𝐬 Following my previous post around Compagny DATA Ingestion on Bronze/Raw layers https://lnkd.in/eK_Q9a9F Some ideas based on my DATA experiences to maybe improve the last SQL on your architecture plan 🚀𝐋𝐨𝐚𝐝 𝐞𝐟𝐟𝐢𝐜𝐢𝐞𝐧𝐭𝐥𝐲 𝐚𝐧𝐝 𝐪𝐮𝐢𝐜𝐤𝐥𝐲 𝐒𝐐𝐋 𝐋𝐇 (𝐠𝐨𝐥𝐝) 𝐨𝐫 𝐃𝐖𝐇 (𝐠𝐨𝐥𝐝) 𝐭𝐚𝐛𝐥𝐞𝐬 𝐏𝐫𝐨𝐯𝐢𝐝𝐞 𝐚 𝐧𝐢𝐜𝐞 𝐒𝐐𝐋 𝐝𝐢𝐦𝐞𝐧𝐬𝐢𝐨𝐧𝐚𝐥 (𝐊𝐢𝐦𝐛𝐚𝐥𝐥) 𝐦𝐨𝐝𝐞𝐥 𝐭𝐨 𝐏𝐨𝐰𝐞𝐫 𝐁𝐈 (𝐚𝐧𝐚𝐥𝐲𝐭𝐢𝐜𝐬 𝐜𝐮𝐛𝐞) 𝐚𝐬 𝐃𝐨𝐦𝐚𝐢𝐧 𝐃𝐀𝐓𝐀 𝐏𝐑𝐎𝐃𝐔𝐂𝐓 Big deal : Challenge on these stacks & last SQL step with strong DATA champion editor : 𝐃𝐁𝐓 , 𝐁𝐢𝐠𝐪𝐮𝐞𝐫𝐲 , 𝐒𝐧𝐨𝐰𝐟𝐥𝐚𝐤𝐞  We are on DATA ANALYSTS garden , they would like something very efficient and rich 𝐇𝐨𝐩𝐢𝐧𝐠 𝐜𝐨𝐦𝐢𝐧𝐠 𝐬𝐨𝐨𝐧 𝐬𝐨𝐦𝐞𝐭𝐡𝐢𝐧𝐠 𝐥𝐢𝐤𝐞 𝐃𝐁𝐓 𝐟𝐨𝐫 "𝐊𝐢𝐦𝐛𝐚𝐥𝐥" 𝐢𝐧𝐝𝐮𝐬𝐭𝐫𝐢𝐚𝐥 𝐭𝐫𝐚𝐧𝐬𝐟𝐨𝐫𝐦𝐚𝐭𝐢𝐨𝐧𝐬 𝐚𝐧𝐝 𝐃𝐐𝐂 - 𝐃𝐚𝐭𝐚 𝐪𝐮𝐚𝐥𝐢𝐭𝐲 𝐜𝐨𝐧𝐭𝐫𝐨𝐥𝐬 -  - 𝐃𝐃𝐋 𝐃𝐞𝐩𝐥𝐨𝐲𝐦𝐞𝐧𝐭𝐬 -𝐃𝐌𝐋 𝐑𝐮𝐧𝐬 - 𝐃𝐨𝐜𝐮𝐦𝐞𝐧𝐭𝐚𝐭𝐢𝐨𝐧 𝐇𝐨𝐰 𝐭𝐨 𝐬𝐚𝐲 ? , 𝐈 𝐥𝐢𝐤𝐞 𝐓 𝐒𝐐𝐋 𝐒𝐭𝐨𝐫𝐞𝐝 𝐩𝐫𝐨𝐜 & 𝐧𝐨𝐭𝐞𝐛𝐨𝐨𝐤 𝐚𝐧𝐝 𝐈 𝐚𝐝𝐦𝐢𝐭 𝐭𝐡𝐚𝐭 𝐭𝐡𝐞𝐲 𝐝𝐨 𝐭𝐡𝐞 𝐣𝐨𝐛 𝐛𝐮𝐭 𝐢𝐭 𝐢𝐬 𝐯𝐞𝐫𝐲 𝐟𝐚𝐫 𝐟𝐫𝐨𝐦 𝐃𝐁𝐓 𝐩𝐫𝐨𝐝𝐮𝐜𝐭𝐢𝐯𝐢𝐭𝐲 𝐚𝐧𝐝 𝐞𝐚𝐬𝐞 𝐨𝐟 T SQL functions on Fabric DWH needs to be very rich Not on the chart , sorry ==> If we want to hide some RAW columns & rows , we need perhaps , like on Bigquery to offer SQL Authorized views (to RAW DWH data) for data analysts (helping load gold tables) Goal : Reduce time to market on build sessions, Kimball transformation to be industrial but simple , improve maintenance , readability , performances 5) [Rows snapshot based primary key and Type2 columns changes] => provide automatically start / end row version datetime , surrogate key 6) [Merge based primary keys and Type 1 columns updates] <== Remark : Perform a real merge ( Fabric LH & TSQL DWH need to accept the famous merge) 5) & 6) will help to easily manage the slow change dimension table by merging on a same SQL view 😁 𝐌𝐀𝐍𝐃𝐀𝐓𝐎𝐑𝐘 𝐎𝐍 𝐀 𝐌𝐎𝐃𝐄𝐑𝐍 𝐃𝐀𝐓𝐀 𝐏𝐥𝐚𝐭𝐟𝐨𝐦 ==> 𝐄𝐯𝐞𝐫𝐲𝐭𝐡𝐢𝐧𝐠 𝐧𝐞𝐞𝐝𝐬 𝐭𝐨 𝐛𝐞 𝐂𝐈𝐂𝐃 & 𝐝𝐚𝐭𝐚 𝐜𝐚𝐭𝐚𝐥𝐨𝐠 𝐜𝐨𝐦𝐩𝐥𝐢𝐚𝐧𝐭 𝐚𝐧𝐝 𝐰𝐞 𝐧𝐞𝐞𝐝 𝐭𝐨 𝐠𝐞𝐭 𝐚𝐧 𝐢𝐧𝐭𝐞𝐠𝐫𝐚𝐭𝐞𝐝 𝐬𝐭𝐫𝐨𝐧𝐠 𝐝𝐚𝐭𝐚 𝐪𝐮𝐚𝐥𝐢𝐭𝐲 𝐬𝐞𝐫𝐯𝐢𝐜𝐞 𝐨𝐧 𝐅𝐚𝐛𝐫𝐢𝐜 (𝐒𝐐𝐋 𝐚𝐧𝐝 𝐏𝐛𝐢 𝐝𝐚𝐭𝐚) 𝐚𝐫𝐨𝐮𝐧𝐝 𝐜𝐨𝐦𝐩𝐥𝐞𝐭𝐞𝐧𝐞𝐬𝐬 , 𝐫𝐮𝐥𝐞𝐬 , 𝐩𝐫𝐨𝐟𝐢𝐥𝐥𝐢𝐧𝐠 Coming soon : Ideas around improve TESTS ( non regression , best practices) on cicd , Power BI workload #data #dataanalytics #datagovernance #powerbi #microsoftfabric #bi #azure #microsoft #DataTransformation #DataCulture #BusinessExperts #DataAwareness #DataMesh #DataCatalog #dataenginners #microsoft #datafactory #dataengineering

4

2 commentaires

𝐒𝐜𝐡𝐞𝐦𝐚 𝐄𝐧𝐟𝐨𝐫𝐜𝐞𝐦𝐞𝐧𝐭 & 𝐄𝐯𝐨𝐥𝐮𝐭𝐢𝐨𝐧 𝐢𝐧 𝐃𝐞𝐥𝐭𝐚 𝐋𝐚𝐤𝐞. 👉 𝐒𝐜𝐡𝐞𝐦𝐚 𝐄𝐧𝐟𝐨𝐫𝐜𝐞𝐦𝐞𝐧𝐭:  - Suitable for scenarios where the source data schema is static. - Ensures that the source data being written into the target table matches the target table definition, thus providing clean data ready for analytics. - Delta Lake uses schema validation on 𝒘𝒓𝒊𝒕𝒆. This means schema compatibility is checked during the insert operation. - If a row's schema doesn't match the target table schema, the data is rejected entirely, and an exception is raised. - The table schema is not changed unless it is intentionally modified. 👉 𝐒𝐜𝐡𝐞𝐦𝐚 𝐄𝐯𝐨𝐥𝐮𝐭𝐢𝐨𝐧: - Suitable for scenarios where the source data schema is dynamic and you intend to change the table's schema automatically. - Allows users to easily change a table's current schema to accommodate data that changes over time. - Activated by adding .option('mergeSchema', 'true'). - Any columns that are in the source DataFrame but not in the target table are automatically added at the end of the target table definition. 👨‍💻 Below references helps you dive deeper on these concepts.  - https://lnkd.in/gWKx2WVD - https://lnkd.in/gsH4U7_w #datalake #deltalake #schema #databricks #deltatables #dataengineering

7

Running multiple models in #dbt Did you know you can use a wildcard to run multiple models based on their name? dbt run --models models/marts/dim_* Will run all the dimensions in my marts layer, if you add to this the '+' option you can tell it to run all the dimensions and all their required dependencies. dbt run --models +models/marts/dim_* Another reason to have a sensible naming convention on your models 😉

3