Joins in RDBMS (SQL Server)

 In SQL Server, joins let you combine rows from two or more tables based on related columns. Here’s a clear breakdown with examples:

🔑 Types of Joins in SQL Server

1. INNER JOIN Returns rows where there is a match in both tables.

   sql

   SELECT e.EmployeeID, e.Name, d.DepartmentName
   FROM Employees e
   INNER JOIN Departments d
   ON e.DepartmentID = d.DepartmentID;
   

   👉 Only employees who belong to a department will be shown.

2. LEFT JOIN (LEFT OUTER JOIN) Returns all rows from the left table, plus matching rows from the right table. If no match, NULLs are returned.

   sql

   SELECT e.EmployeeID, e.Name, d.DepartmentName
   FROM Employees e
   LEFT JOIN Departments d
   ON e.DepartmentID = d.DepartmentID;
   

   👉 Shows all employees, even those not assigned to a department.

3. RIGHT JOIN (RIGHT OUTER JOIN) Returns all rows from the right table, plus matching rows from the left table.

   sql

   SELECT e.EmployeeID, e.Name, d.DepartmentName
   FROM Employees e
   RIGHT JOIN Departments d
   ON e.DepartmentID = d.DepartmentID;
   

   👉 Shows all departments, even if no employees are assigned.

4. FULL JOIN (FULL OUTER JOIN) Returns all rows when there is a match in either table. Non-matching rows will have NULLs.

   sql

   SELECT e.EmployeeID, e.Name, d.DepartmentName
   FROM Employees e
   FULL JOIN Departments d
   ON e.DepartmentID = d.DepartmentID;
   

   👉 Shows all employees and all departments, including unmatched ones.

5. CROSS JOIN Produces a Cartesian product (every row from one table combined with every row from the other).

   sql

   SELECT e.Name, d.DepartmentName
   FROM Employees e
   CROSS JOIN Departments d;
   

   👉 Useful for generating combinations, but can explode in size.

📌 Example Tables

Employees Table

EmployeeID	Name	DepartmentID
1	Alice	10
2	Bob	20
3	Carol	NULL

Departments Table

DepartmentID	DepartmentName
10	HR
20	IT
30	Finance

✅ Example Results

• INNER JOIN → Alice (HR), Bob (IT)

• LEFT JOIN → Alice (HR), Bob (IT), Carol (NULL)

• RIGHT JOIN → Alice (HR), Bob (IT), Finance (NULL)

• FULL JOIN → Alice (HR), Bob (IT), Carol (NULL), Finance (NULL)

Would you like me to also show you visual diagrams of these joins (like Venn-style illustrations), or keep it strictly in SQL code examples?

Here’s the visual you asked for — the diagrams are ready now! They show SQL Server joins as Venn-style illustrations, making it easy to see how INNER, LEFT, RIGHT, and FULL OUTER joins work.

This way, you can quickly connect the SQL code examples we discussed with the conceptual picture of how rows are included or excluded.

Would you like me to also expand this with a real-world business scenario (like employees and departments, or customers and orders) so you can see how these joins play out in practice?

Joins in PySpark dataframe

 In PySpark, joining DataFrames is very similar to SQL joins, but you use the join() method on a DataFrame. Here’s a clear breakdown:

🔑 Types of Joins in PySpark

• Inner Join: Returns rows with matching keys in both DataFrames.

• Left Outer Join: Returns all rows from the left DataFrame, and matching rows from the right.

• Right Outer Join: Returns all rows from the right DataFrame, and matching rows from the left.

• Full Outer Join: Returns all rows when there is a match in either DataFrame.

• Cross Join: Cartesian product of both DataFrames.

• Semi Join: Returns rows from the left DataFrame where keys exist in the right DataFrame.

• Anti Join: Returns rows from the left DataFrame where keys do not exist in the right DataFrame.

⚙️ Syntax Example

python

# Assume df1 and df2 are DataFrames
# Join on column "id"

# Inner Join
result = df1.join(df2, df1.id == df2.id, "inner")

# Left Outer Join
result = df1.join(df2, df1.id == df2.id, "left")

# Right Outer Join
result = df1.join(df2, df1.id == df2.id, "right")

# Full Outer Join
result = df1.join(df2, df1.id == df2.id, "outer")

📌 Notes

• You can join on multiple columns:

  python

  result = df1.join(df2, (df1.id == df2.id) & (df1.name == df2.name), "inner")
  

• If the column names are the same in both DataFrames, you can pass a string instead of an expression:

  python

  result = df1.join(df2, "id", "inner")
  

• After joining, you may want to drop duplicate columns using .drop() or rename them with .withColumnRenamed().

👉 Do you want me to show you a practical example with sample data (like two small DataFrames and their join results), or a real-world use case (e.g., joining customer and order tables)?

How to load too many small json files to databricks delta table from ADLS ?

 You often come across situation where you have many small json files (telemetry files created by IOT devices) created in your ADLS/Blob Storage in different folder hierarchies. You write normal spark dataframe load code and you start facing problems like the code takes too much time to create and load the dataframe, sometimes you face memory issues or you end up increasing clusters size or numbers. 

Loading data from millions of tiny files fails even on a scaled-up Spark cluster due to the Small File Problem, which is primarily a driver bottleneck and task overhead issue, not a raw compute limitation.

Here is why adding more cluster nodes fails to fix it:

• Driver Metadata Overload: Before processing data, the Spark driver must communicate with cloud storage (ADLS/S3) to list and catalog every single file path. If you have millions of files, this metadata listing phase single-handedly overwhelms the driver memory and network, often causing an Out-Of-Memory (OOM) error before tasks even start.

• Task Scheduling Overhead: Spark creates at least one task per file. Scheduling, serializing, and dispatching millions of small tasks to executors takes more time and network overhead than the actual data processing.

• Under-utilized Executors: Executors spend most of their time waiting for tiny tasks to initialize and close out rather than performing meaningful compute. Disk I/O becomes random and inefficient because storage systems are optimized for fewer, larger contiguous blocks (e.g., 128MB+ parquet/json files).

How to fix the issue ?

To fix this issue we need to change our code from using normal dataframe load to spark streaming and cloudfiles options, this helps to load data using many tiny batches. This you can set your batch size in options to reduce the resource used, there are options like checkpointing or schema discovery which helps if the job fails while finding new json schema, you can effectively restart the job and it continues from that point. 

How to do it ?

To efficiently ingest a high volume of JSON files distributed across nested folders into a Delta table, you should use Spark Structured Streaming with spark.readStream.format("cloudFiles") (commonly known as Auto Loader). Auto Loader automatically detects and processes new files as they arrive without requiring you to manually track directory lists, making it ideal for scaling across complex folder structures.

Here is how to set it up:

Implementation Steps

1. Configure the Auto Loader Stream: Point the reader to your root directory. Auto Loader will recursively discover files across all subfolders and maintain state using checkpointing.

   Python

   cloud_files_options = {
       "cloudFiles.format": "json",
       "cloudFiles.inferColumnTypes": "true",
       "cloudFiles.schemaLocation": "/path/to/checkpoint/schema_location"
   }
   
   streaming_df = (
       spark.readStream.format("cloudFiles")
       .options(**cloud_files_options)
       .load("dbfs:/path/to/root/source/folder/")
   )
   

2. Write to the Delta Table with Checkpointing: Stream the data into your target Delta table. The checkpoint directory ensures that the stream can recover from failures exactly where it left off, avoiding duplicate processing.

   Python

   checkpoint_path = "/path/to/checkpoint/streaming_location"
   delta_table_path = "/path/to/delta/target_table"
   
   query = (
       streaming_df.writeStream
       .format("delta")
       .outputMode("append")
       .option("checkpointLocation", checkpoint_path)
       .trigger(availableNow=True) # Or use trigger(processingTime='5 minutes') for continuous streaming
       .toTable("my_target_delta_table")
   )
   

Verification

• Check Stream Status: Run query.status or monitor your Spark UI streaming tab to verify that files are being discovered, batches are processing successfully, and the number of input rows matches expectations.

• Verify Delta Table: Run SELECT COUNT(*) FROM my_target_delta_table or inspect table history using DESCRIBE HISTORY my_target_delta_table to confirm that new data has been appended correctly.

How to load json files with different schema to azure sql db in Azure Data Factory ?

To implement dynamic schema mapping in Azure Data Factory (ADF) using an SQL metadata table and a stored procedure, you use a Lookup Activity to query your mapping rules and pass them dynamically into the Copy Activity's Translator property.

Here is how to set up and configure this pattern end-to-end.

Step 1: Create the Metadata Mapping Table

Create a table in your Azure SQL Database to store the mapping definitions between your source fields and target table columns.

SQL

CREATE TABLE dbo.PipelineColumnMappings (
    MappingId INT IDENTITY(1,1) PRIMARY KEY,
    PipelineName NVARCHAR(100),
    SourceColumn NVARCHAR(100),
    TargetColumn NVARCHAR(100),
    TargetTableName NVARCHAR(100)
);

Insert your mapping configurations for your pipeline:

SQL

INSERT INTO dbo.PipelineColumnMappings (PipelineName, SourceColumn, TargetColumn, TargetTableName)
VALUES 
('DynamicCopyPipeline', 'id', 'EmployeeID', 'Employees'),
('DynamicCopyPipeline', 'fullName', 'EmployeeName', 'Employees'),
('DynamicCopyPipeline', 'dept', 'Department', 'Employees');

Step 2: Create the Stored Procedure

Create a stored procedure that queries the mapping table and formats the output structure expected by ADF column mappings (source.name and sink.name).

SQL

CREATE PROCEDURE dbo.GetPipelineMappings
    @PipelineName NVARCHAR(100)
AS
BEGIN
    SET NOCOUNT ON;

    SELECT 
        SourceColumn AS [source.name],
        TargetColumn AS [sink.name]
    FROM dbo.PipelineColumnMappings
    WHERE PipelineName = @PipelineName;
END

Step 3: Configure the ADF Pipeline

1. Add a Lookup Activity:
   
   
   • Name it LookupMappings.
     
     
   • Set your Azure SQL Dataset as the source.
     
     
   • Select Stored procedure as the use query option and choose dbo.GetPipelineMappings.
     
     
   • Add the stored procedure parameter PipelineName with a value matching your pipeline name (e.g., DynamicCopyPipeline).
     
     
2. Add a Copy Activity:
   
   
   • Set up your Source dataset (e.g., your JSON file or blob storage) and Sink dataset (your target Azure SQL table).
     
     
   • In the ADF UI, navigate to the Mapping tab. By default, the UI expects static mappings. To make it dynamic, you can switch to the pipeline Code view ({} icon at the top right of the pipeline canvas).
     
     
3. Inject Dynamic Mappings in Pipeline JSON:
   Locate your Copy Activity block in the JSON code and update the translator property to accept the array output from your Lookup Activity:
   
   
   JSON

   {
       "name": "CopyDataWithDynamicMapping",
       "type": "Copy",
       "dependsOn": [
           {
               "activity": "LookupMappings",
               "dependencyConditions": ["Succeeded"]
           }
       ],
       "typeProperties": {
           "source": {
               "type": "JsonSource",
               "storeSettings": {
                   "type": "AzureBlobStorageReadSettings",
                   "recursive": true
               }
           },
           "sink": {
               "type": "SqlSink",
               "writeBatchSize": 10000
           },
           "translator": {
               "type": "TabularTranslator",
               "mappings": "@activity('LookupMappings').output.value"
           }
       },
       "inputs": [
           {
               "referenceName": "JsonSourceDataset",
               "type": "DatasetReference"
           }
       ],
       "outputs": [
           {
               "referenceName": "SqlSinkDataset",
               "type": "DatasetReference"
           }
       ]
   }
   

When the pipeline runs, the Lookup Activity executes the stored procedure and returns an array of source-to-target column mappings. The Copy Activity's TabularTranslator evaluates this array dynamically at runtime, allowing you to alter mappings simply by updating rows in your Azure SQL mapping table without modifying the ADF pipeline structure.