Satheesh Kannan

Latest blog on how to connect Oracle Autonomous Data Warehouse (ADW) to Apache Superset using the python-oracledb thick client on an Oracle Cloud Infrastructure (OCI) compute instance. Connection between Oracle ADW and Apache Superset can be quite the maze, given the scattered pieces of information out there. This step-by-step guide brings all the necessary details together in one place, making it much easier for you to set up and start exploring your data without much hassle. Thanks to @christopher jones (https://github.com/cjbj). His insights have been helpful for this. https://lnkd.in/g6h5Fi5u #Superset #PythonOracledb #OracleCloud

10

Here's my latest article on Medium for various aspects of Snowflake Technology Infrastructure which needs to be managed and governed . I will be diving into each of these aspects in my upcoming articles . Please feel free to subscribe , share and comment. Reach out for any professional help or consulting services in these areas . #snowflakecm #snowflakeintrastructure #snowflakedevops

7

SPARK code to find the difference in price of a commodity, from its next day price. #SCALA #SPARK #BIGDATA Karthik K. def main(args: Array[String]): Unit = { val ss = SparkSession.builder.master("local[*]").appName("DFDEMO").getOrCreate() import ss.implicits._ val wind = Window.orderBy(col("PRICE_DATE")) 👈 //CREATING WINDOW val df_1 = df.select(col("PRICE"), lead(col("PRICE"), 1).over(wind).as("NEXT_PRICE")) val df_2 = df_1.withColumn("DIFF_IN_PRICE", (col("NEXT_PRICE") - col("PRICE"))) 👈 //NEW COLUMN TO CAPTURE DIFFERENCE IN PRICES 👇 //CREATING NEW COLUMN TO CAPTURE REMARKS df_2.select(col("PRICE"),col("NEXT_PRICE"),col("DIFF_IN_PRICE"),when(col("DIFF_IN _PRICE")===0,"NO CHANGE"). when(col("DIFF_IN_PRICE") >0,"INCREMENT").when(col("DIFF_IN_PRICE") <0,"DECREMENT"). otherwise("EXCEPTION").as("REMARKS")).show() OUTPUT:

6

SPARK code to find out the product, which generates the maximum revenue for a commercial store. #SCALA #SPARK #BIGDATA Karthik K. INPUT: PRODUCT_ID,QUANTITY_SOLD,PRICE 101,10,150 101,15,150 102,5,50 102,10,50 103,3,100 103,10,100 QUERY: def main(args: Array[String]):Unit = { val ss = SparkSession.builder.appName("two_dfS").master("local[*]").getOrCreate() val df_1 = ss.read.format("csv").option("header",true). option("path","C:/Users/hp/PRODUCT_ID.txt"). schema("PRODUCT_ID Int, QUANTITY_SOLD Int, PRICE Int").load() val df_2 = df_1.withColumn("REVENUE",col("QUANTITY_SOLD")*col("PRICE")) val df_3 =df_2.groupBy(col("PRODUCT_ID")). agg(sum(col("QUANTITY_SOLD")).as("TOTAL_QUANTITY_SOLD"), sum(col("REVENUE")).as("TOTAL_REVENUE")) df_3.show() val wind = Window.orderBy(col("TOTAL_REVENUE").desc) val df_4 = df_3.select(col("PRODUCT_ID"),col("TOTAL_REVENUE"),row_number().over(wind).as("ROW_NUMBER")) df_4.filter(col("ROW_NUMBER")<=1).show() OUTPUT:

4

🚀 Optimise INSERT, UPDATE and DELETE in PostgreSQL using best indexing strategy! 📊 Struggling with sluggish DML operations due to indexes? Refer my new medium blog https://lnkd.in/gTxmyWaY which elaborates best indexing strategies to boost your PostgreSQL performance, especially for tables in the TB range. 📈 1. Evaluate and Refine Index Types:    - 🧹 Review Index Usage: Identify and drop unused indexes using `pg_stat_all_indexes`.    - 🔍 Choose the Right Index Type: Use B-tree for high-cardinality columns and BRIN for range queries on naturally sorted columns.    - ✂️ Partial Indexes: Index only frequently queried subsets to reduce size and improve speed.    - 📚 Covering Indexes: Ensure indexes cover common queries to minimize extra lookups. 2. Remove Redundant Indexes:    - 🗑️ Unused indexes can slow down operations. Regularly review and drop them to save space and enhance performance. 3. Advanced Index Types:    - ⚡ Expression Indexes: Index results of expressions or functions for efficient querying based on calculations.    - 🧩 Partial Indexes: Optimize for specific data subsets to reduce index size. 4. Use Efficient Index Types:    - 🏷️ BRIN Indexes: Ideal for large, naturally sorted tables. Efficient for range queries.    - 🔢 HASH Indexes: Perfect for exact-match queries on large datasets. 5. Reorganize Indexes:    - 🔄 Reindex During Off-Peak Hours: Schedule maintenance to minimize impact.    - ⚙️ REINDEX CONCURRENTLY: Reduce downtime by using PostgreSQL's concurrent reindexing. 6. Consider Index Maintenance Tools:    - 🛠️ pg_repack: Reorganize indexes without locking tables using this powerful extension.    - 🌐 Logical Replication: Perform maintenance on a replica, then promote it to primary. 7. Tune Configuration Parameters:    - 🔧 Increase maintenance_work_mem: Speed up index operations by allocating more memory. By following these strategies, you can ensure your PostgreSQL database performs optimally, even with massive datasets. Stay tuned for more tips on table structure optimization, partitioning, and sharding in our future posts! 🚀📈 #SQL #DatabaseOptimization #PostgreSQL #DataManagement #TechTips

44

Spark Query to create a new column based on age column in the table. #BIGDATA #SCALA #SPARK Karthik K. def main(args:Array[String]):Unit={ val ss = SparkSession.builder.appName("WHEN_OTHERWISE").master("local[*]").getOrCreate() import ss.implicits._ val data = List(("M",17),("F",25),("M",30),("F",35),("M",10)).toDF("Gender","Age") val df_with_age_remark = data.select(col("Gender"),col("Age"),when(col("Age")>=18,true).otherwise(false).as("IS ADULT")) println("Printing Age Remarks") df_with_age_remark.show() OUTPUT:

2

Snowflake summit 2024 highlights doc for your reference. #snowflake #summit #updates

49

There are multiple ways in which Snowflake can handle Data deduplication for you. In my latest article on medium, I did describe few of those options. Feel free to share, comment and subscribe. #snowflakeadvocate #snowflakedata #snowflakenineering

5

What is Delta lake and how is it different that other data storage frameworks? Delta Lake is the optimized storage layer that provides the foundation for tables in a lakehouse on Databricks. It enables organizations to access and analyze new data in real time. Data warehouses excel in structured data and fast queries but falter when confronted with unstructured data. Conversely, data lakes adeptly handle unstructured data but lack the performance needed for quick queries. Delta lakes combines the best of both data warehouse and data lake. What are the features and benefits of Delta Lake? 1️⃣ Open Format Compatibility: Delta Lake leverages Apache Parquet and seamlessly integrates with Apache Spark for flexible operations. 2️⃣ ACID Transactions: Delta Lake ensures data integrity with ACID properties, providing accurate audit trails. 3️⃣ Time Travel: Delta Lake's transaction log enables precise data recreation at any point in time. 4️⃣ Schema Enforcement: Delta Lake enforces data consistency, preventing corruption and enhancing reliability. 5️⃣ Data Manipulation Language (DML) Operations: Delta Lake supports merge, update, and delete commands for complex use cases. Which platforms support Delta Lake? It works seamlessly with popular platforms like Databricks, Apache Spark, Azure Synapse Analytics, AWS Glue, and Google Cloud Platform, making data management easier and more flexible for users everywhere In summary, Delta Lake transforms your data lake into a more reliable and performant data platform. It ensures data quality, supports complex data processing needs, and enhances overall performance. #dataengineering #WhatsTheData

24

Converting a list to string, with a delimiter. Implementation of inbuilt mkString() function in SCALA. #SCALA #SCALADEVELOPER #BIGDATADEVELOPER #BIGDATA Karthik K. def main(args: Array[String]): Unit ={ //var a = "Tanuj" //lowerCase(a) var s = "" var l = List(1,2,3,4,5,6,7,8) for( i <- 0 until l.length) { if( i == l.length -1) { s = s.concat(l(i).toString) } else { s = s.concat(l(i).toString) s = s.concat(':'.toString) } } print("The elements of list when converted to string are:-" + s) } } OUTPUT:

3

Consider a DataFrame named sales with the following columns: product_id, quantity_sold, and revenue. Write Spark DataFrame code to find the total quantity sold and revenue for each product. #SCALA #SPARK #BIGDATA Karthik K. INPUT: PRODUCT_ID,QUANTITY_SOLD,PRICE 101,10,150 101,15,150 102,5,50 102,10,50 103,3,100 103,10,100 QUERY: def main(args: Array[String]):Unit = { val ss = SparkSession.builder.appName("two_dfS").master("local[*]").getOrCreate() val df_1 = ss.read.format("csv").option("header",true). option("path","C:/Users/hp/PRODUCT_ID.txt"). schema("PRODUCT_ID Int, QUANTITY_SOLD Int, PRICE Int").load() val df_2 = df_1.withColumn("REVENUE",col("QUANTITY_SOLD")*col("PRICE")) df_2.groupBy(col("PRODUCT_ID")). agg(sum(col("QUANTITY_SOLD")).as("TOTAL_QUANTITY_SOLD"), sum(col("REVENUE")).as("TOTAL_REVENUE")).show() OUTPUT:

3

Spark can be challenging when it comes to optimizing write operations. This post really helps visualize and explain the concepts.

8

1 Comment

Snowflake Optimization Tip of the Day #5 ➡ Use incremental models in dbt There is no simpler way to reduce Snowflake cost than to process less data. If you are using dbt, try to write every model incrementally. Instead of reprocessing all data every time, incremental models only process new or changed data since the last run. This is achieved primarily by using date stamps or other markers to identify what data has been added or modified since the last run. Full-refreshes are very expensive and should be avoided if possible.

8

Here's my latest article on Medium about a very small but key Design aspect of any Data Practitioner's journey, data checksum. #snowflake #snowflakeadvocate #checksum

3

Was the #acquisition of Tabular (now part of Databricks) a strategic move by Databricks? Although the delta table format is good in many ways, the #iceberg format excelled in solving some major problems in datalake design. In last couple of years #Apache #Iceberg has definitely won the hearts of Data engineers and data architects. In my opinion Iceberg was starting to become a major threat to Delta format and Tabular was a major contributor to Apache iceberg project. Hence, as a strategic move did Databricks decide to win over the competitor by buying out the competition? This seems like a strategic move as laid out in the 3rd chapter of #artofwar by Sun Tzu. “Sun Tzu said: In the practical art of war, the best thing of all is to take the enemy's country whole and intact; to shatter and destroy it is not so good. So, too, it is better to capture an army entire than to destroy it, to capture a regiment, a detachment, or a company entire than to destroy them.” What does this mean to iceberg community? What are your thoughts? #TheArtofWar #suntzu #apacheiceberg

2

🗓 Snowflake Optimization - Summary of the week 1. Prune Partitions Snowflake uses the concept of micropartitions. Apply filters to prune data. 2. Enable Automatic Clustering Automatic clustering can be useful for certain tables that are heavily queried on certain fields. It does cost credits, so you might need to a cost-benefit analysis 3. Reduce Idle time Set AUTO_SUSPEND = 60. 4. Consolidate warehouses Consolidate similar warehouses into a single multicluster warehouse. 5. Use incremental models Write incremental models in dbt. Incremental models only process the new incoming data based on a field, generally date.

21

1 Comment

Questions on COALESCE in SPARK #BIGDATA #SPARK Karthik K. 𝐖𝐡𝐚𝐭 𝐢𝐬 𝐭𝐡𝐞 𝐮𝐬𝐞 𝐨𝐟 𝐜𝐨𝐚𝐥𝐞𝐬𝐜𝐞 𝐢𝐧 𝐒𝐩𝐚𝐫𝐤? In Apache Spark, coalesce is a transformation that is used to reduce the number of partitions in an RDD (Resilient Distributed Dataset) or DataFrame. This is particularly useful for optimizing the performance of certain operations by reducing the shuffling of data across the network. 𝐖𝐡𝐲 𝐔𝐬𝐞 𝐜𝐨𝐚𝐥𝐞𝐬𝐜𝐞? 𝐎𝐩𝐭𝐢𝐦𝐢𝐳𝐢𝐧𝐠 𝐏𝐞𝐫𝐟𝐨𝐫𝐦𝐚𝐧𝐜𝐞: By reducing the number of partitions, coalesce can help optimize performance for certain operations, such as actions or joins, by minimizing the overhead of managing too many partitions. 𝐀𝐯𝐨𝐢𝐝𝐢𝐧𝐠 𝐅𝐮𝐥𝐥 𝐒𝐡𝐮𝐟𝐟𝐥𝐞: Unlike repartition, coalesce avoids a full shuffle of the data. This is particularly useful when reducing the number of partitions without significantly changing the data distribution. 𝐂𝐨𝐝𝐢𝐧𝐠 𝐐𝐮𝐞𝐬𝐭𝐢𝐨𝐧𝐬 𝐟𝐨𝐫 𝐈𝐧𝐭𝐞𝐫𝐯𝐢𝐞𝐰 𝐁𝐚𝐬𝐢𝐜 𝐔𝐧𝐝𝐞𝐫𝐬𝐭𝐚𝐧𝐝𝐢𝐧𝐠: 1] What is the purpose of coalesce in Spark? 2] How does coalesce differ from repartition? 𝐏𝐫𝐚𝐜𝐭𝐢𝐜𝐚𝐥 𝐒𝐜𝐞𝐧𝐚𝐫𝐢𝐨: 1] Given a DataFrame with 10 partitions, how would you reduce it to 3 partitions without causing a full shuffle? 2] What are the potential benefits of reducing the number of partitions using coalesce before performing an action like count? 𝐏𝐞𝐫𝐟𝐨𝐫𝐦𝐚𝐧𝐜𝐞 𝐂𝐨𝐧𝐬𝐢𝐝𝐞𝐫𝐚𝐭𝐢𝐨𝐧𝐬: 1] Explain a scenario where using coalesce would significantly improve performance. 2 ]What are the implications of using coalesce on data skew and load balancing?

9

SPARK code to impute NULL or missing values in a table containing NULL in multiple columns. #SPARK #SCALA #BIGDATA Karthik K. def main(args:Array[String]):Unit = { val ss = SparkSession.builder.master("local[*]").appName("DFDEMO").getOrCreate() import ss.implicits._ val df_1 = ss.read.format("csv").option("header",true). option("path","C:/Users/hp/COMPANY.txt").load() df_1.show() val df_2 = df_1.select(col("PRODUCT_NAME"),col("PRODUCT_ID") , (when(col("PRODUCT_ID").isNull,0).otherwise(col("PRODUCT_ID")).as("LEGIT_PRODUCT_ID")), when(col("PRODUCT_NAME").isNull,"UNKNOWN COMPANY"). otherwise(col("PRODUCT_NAME")).as("LEGIT_PRODUCT_NAME")) df_2.show() OUTPUT:

26

4 Comments

How the Sort Merge Join Happens in the Apache Spark , Know the SMJ internals Sort-Merge Join (SMJ): In Apache Spark, Sort-Merge Join (SMJ) is the default join strategy for combining datasets based on a shared join key.  It excels at handling large datasets that might not entirely reside in memory on a single worker node. its execution phases: 1. Shuffle Phase: Spark partitions both input DataFrames according to the join key. This ensures records with matching join keys end up on the same worker node. Each partition is shuffled and sent across the network to the appropriate worker nodes based on the join key's hash value. 2. Sort Phase: Within each worker node, the shuffled partitions are sorted independently and in parallel using a stable sorting algorithm (like Timsort) to maintain the order of join keys. 3. Merge Phase: Worker nodes iterate through their sorted partitions simultaneously, performing a merge join operation. They compare the join keys of each DataFrame, and when a match is found, they combine the corresponding rows into a new joined row. This process continues until all matching join keys from both DataFrames have been processed. Example: Imagine you have two DataFrames: customers: (customer_id, name, city) orders: (order_id, customer_id, amount) You want to join them on the customer_id to get customer details along with their orders. Execution Steps: Shuffle: Both customers and orders are partitioned and shuffled based on customer_id. Sort: On each worker node, the shuffled partitions for customers and orders are sorted independently by customer_id. Merge: Worker nodes iterate through their sorted partitions, comparing customer_id values. When a match is found, a new row is created with data from both DataFrames (e.g., (customer_id, name, city, order_id, amount)). Key Points: SMJ works well for datasets that don't fit entirely in memory or where the join key has good cardinality (distinct values). It requires shuffling data, which can be network-intensive. Spark can automatically choose a more efficient join strategy like broadcast join if one DataFrame is significantly smaller than the other. Additional Considerations: You can configure the number of shuffle partitions using spark.sql.shuffle.partitions. SMJ can be optimized by using in-memory sorting if data fits in memory (spark.sql.sortMergeJoin.forceInMemory=true). For very large datasets, Spark might spill intermediate results to disk during shuffling and sorting, potentially impacting performance. TrendyTech #Spark, #BigData, #DataEngineering, #SoftwareEngineering

37