what is lambda function in python and spark

 A lambda function, also known as an anonymous function, is a small and unnamed function defined using the `lambda` keyword. It is often used for short-term tasks, such as in functional programming operations like `map`, `filter`, and `reduce`. Here's a quick overview of how lambda functions work in both Python and PySpark:

### Python Lambda Function

A lambda function in Python can take any number of arguments but can only have one expression. The syntax is as follows:

```python
lambda arguments: expression
```

Here’s an example of using a lambda function to add two numbers:

```python
add = lambda x, y: x + y
print(add(2, 3))  # Output: 5
```

Lambda functions are often used with functions like `map()`, `filter()`, and `reduce()`:

```python
# Using lambda with map
numbers = [1, 2, 3, 4, 5]
squared = list(map(lambda x: x ** 2, numbers))
print(squared)  # Output: [1, 4, 9, 16, 25]

# Using lambda with filter
even_numbers = list(filter(lambda x: x % 2 == 0, numbers))
print(even_numbers)  # Output: [2, 4]

# Using lambda with reduce
from functools import reduce
product = reduce(lambda x, y: x * y, numbers)
print(product)  # Output: 120
```

### Lambda Function in PySpark

In PySpark, lambda functions are used in similar ways, especially with operations on RDDs. Here are some examples:

```python
from pyspark import SparkContext

sc = SparkContext("local", "example")

# Creating an RDD
rdd = sc.parallelize([1, 2, 3, 4, 5])

# Using lambda with map
squared_rdd = rdd.map(lambda x: x ** 2)
print(squared_rdd.collect())  # Output: [1, 4, 9, 16, 25]

# Using lambda with filter
even_rdd = rdd.filter(lambda x: x % 2 == 0)
print(even_rdd.collect())  # Output: [2, 4]

# Using lambda with reduce
product_rdd = rdd.reduce(lambda x, y: x * y)
print(product_rdd)  # Output: 120
```

In both Python and PySpark, lambda functions provide a concise and powerful way to perform operations on data, especially in contexts where defining a full function would be overkill.

Factorial of a number

 

Recursive Approach

def factorial(n):
    if n == 0 or n == 1:
        return 1 ;
    else:
        return n * factorial(n - 1)

print(factorial(5))  # Output: 120

Iterative Approach

def factorial(n):
    result = 1
    for i in range(1, n + 1):    -- for and return statements are in same indent
      result *= i
    return result

print(factorial(5))  # Output: 120

sort in python

 

The sort() method in Python is used to sort a list in place, meaning the list itself is modified and reordered. It can sort the list in ascending or descending order based on the values or a custom key function.

---

Syntax:

list.sort(key=None, reverse=False)

Parameters:

1. key (optional):
   • A function that specifies a sorting criterion. By default, elements are sorted based on their natural order.
   • Example: key=len sorts the list by the length of each element.
2. reverse (optional):
   • If True, the list is sorted in descending order. Default is False (ascending order).

---

Examples:

1. Basic Sorting (Ascending Order):

numbers = [3, 1, 4, 1, 5, 9]
numbers.sort()
print(numbers)  # Output: [1, 1, 3, 4, 5, 9]

2. Sorting in Descending Order:

numbers.sort(reverse=True)
print(numbers)  # Output: [9, 5, 4, 3, 1, 1]

3. Custom Sorting Using key:

# Sort strings by their length
words = ["apple", "banana", "cherry", "date"]
words.sort(key=len)
print(words)  # Output: ['date', 'apple', 'banana', 'cherry']

4. Sorting with a Custom Function:

# Sort numbers by their distance from 5
numbers = [10, 2, 8, 3, 6]
numbers.sort(key=lambda x: abs(x - 5))
print(numbers)  # Output: [6, 3, 8, 2, 10]

---

Key Points:

1. In-place Sorting:

   • sort() modifies the original list.
   • If you need a new sorted list without changing the original, use the sorted() function instead.

   original = [3, 1, 4]
   sorted_list = sorted(original)  # New sorted list
   print(original)  # Output: [3, 1, 4]
   

2. Non-comparable Elements:

   • Sorting a list with incompatible types (e.g., numbers and strings) will raise a TypeError.

3. Efficient Sorting:

   • sort() uses the Timsort algorithm, which is highly optimized and stable.

The sort() method is ideal for in-place sorting, while sorted() is more versatile for generating new sorted lists.

Map, Filter, and Reduce in PySpark

 In PySpark, map, filter, and reduce are operations applied to Resilient Distributed Datasets (RDDs) to perform transformations and actions. These operations are foundational for distributed data processing in PySpark.

 

1. Map

The map transformation applies a function to each element of the RDD and returns a new RDD with transformed elements.

 

from pyspark import SparkContext

sc = SparkContext("local", "Map Example")

# Create an RDD
rdd = sc.parallelize([1, 2, 3, 4, 5])

# Use map to square each element
mapped_rdd = rdd.map(lambda x: x ** 2)

print(mapped_rdd.collect())  # Output: [1, 4, 9, 16, 25]
 

2. Filter

The filter transformation selects elements from the RDD that satisfy a given condition and returns a new RDD.

 

# Filter RDD to select only even numbers
filtered_rdd = rdd.filter(lambda x: x % 2 == 0)

print(filtered_rdd.collect())  # Output: [2, 4]
 

 

# Filter RDD to select only even numbers
filtered_rdd = rdd.filter(lambda x: x % 2 == 0)

print(filtered_rdd.collect())  # Output: [2, 4]
 

3. Reduce

The reduce action aggregates the elements of the RDD using a binary operator, returning a single value.

Example:

# Reduce RDD to compute the sum of elements
sum_result = rdd.reduce(lambda x, y: x + y)

print(sum_result)  # Output: 15
 

Combining Map, Filter, and Reduce

These operations can be combined to perform complex computations in a distributed manner.

Example:

# Combine map, filter, and reduce
result = rdd.map(lambda x: x ** 2) \
            .filter(lambda x: x > 10) \
            .reduce(lambda x, y: x + y)

print(result)  # Output: 41 (16 + 25 from squares greater than 10)
 

 

 

Key Points:

1. Lazy Evaluation:

   • map and filter are transformations, so they are lazily evaluated and executed only when an action (e.g., reduce, collect, count) is called.

2. Distributed Nature:

   • These operations are performed in a distributed manner, with map and filter transforming partitions independently, while reduce requires shuffling data across partitions.

3. RDD Focus:

   • These operations work on RDDs. For DataFrames, equivalent operations like select, filter, and agg are more commonly used.

    

   If you're working on large-scale data, consider using PySpark DataFrame APIs for better performance and easier optimization by the Spark Catalyst opti
   

 

 

 

 

 

Where do we use pandas in Data Engg

 

Pandas is extensively used in Data Engineering for various tasks related to data manipulation, cleaning, transformation, and analysis. While Pandas is more commonly associated with Data Analysis, its capabilities make it an essential tool for small-to-medium-scale Data Engineering tasks or prototyping.

Here are specific areas where Pandas is used in Data Engineering:

 

Data Extraction (ETL Process)

• Use Case: Extracting data from various sources such as CSV files, Excel spreadsheets, databases, APIs, and more.
•  
  
• import pandas as pd
  # Load data from CSV
  df = pd.read_csv("data.csv")
  
  # Load data from a SQL database
  import sqlite3
  conn = sqlite3.connect('database.db')
  df = pd.read_sql("SELECT * FROM table_name", conn)
   
  

 2. Data Cleaning

Handling missing values, duplicates, and inconsistent data types.

# Drop rows with missing values
df.dropna(inplace=True)

# Fill missing values with default
df.fillna(value={"column": 0}, inplace=True)

# Remove duplicate rows
df.drop_duplicates(inplace=True)
 

3. Data Transformation

Aggregation, filtering, sorting, and feature engineering.

# Filter rows based on a condition
df_filtered = df[df["age"] > 30]

# Add a new column
df["salary_with_bonus"] = df["salary"] * 1.1

# Grouping and aggregation
df_grouped = df.groupby("department").agg({"salary": "mean"})
 

4. Data Validation and Profiling

 

Ensuring data consistency and profiling the dataset.

# Check for data types
print(df.dtypes)

# Get summary statistics
print(df.describe())

# Validate uniqueness of a column
assert df["id"].is_unique

5. Small-scale Data Pipelines

 # Extract data
df = pd.read_csv("input.csv")

# Transform data
df["total_cost"] = df["quantity"] * df["price"]
df = df[df["total_cost"] > 100]

# Load data to a new file
df.to_csv("output.csv", index=False)

6. Data Integration

 Merging and joining datasets from different sources.

 # Merge two datasets
df_merged = pd.merge(df1, df2, on="id", how="inner")

. Prototyping and Testing

• Rapidly prototyping ETL logic before scaling it to frameworks like PySpark, Dask, or Apache Beam.

 

# Prototype logic using Pandas
df_transformed = df.groupby("category").sum()

# Scale up the logic in PySpark later
 

 # Prototype logic using Pandas
df_transformed = df.groupby("category").sum()

# Scale up the logic in PySpark later

When to Use Pandas in Data Engineering:

1. Small to Medium Datasets: When data fits in memory (up to a few GBs).
2. Prototyping: Quick tests before scaling with distributed systems like PySpark or Dask.
3. Ad-hoc Analysis: One-off or exploratory analysis.
4. Preprocessing: Cleaning and transforming data before loading it into a database or data warehouse.

For larger datasets, distributed computing frameworks like PySpark, Dask, or Apache Beam are often more appropriate.

 

 

 

How to check class of another child class in python

In Python, you can use the isinstance() function to check if an object is an instance of a specific class or any class derived from it. You can also use the issubclass() function to check if a class is a subclass of another class.

 

class Parent:
    pass

class Child(Parent):
    pass

class Grandchild(Child):
    pass

# Creating instances
p = Parent()
c = Child()
g = Grandchild()

# Checking instances
print(isinstance(c, Parent))  # Output: True, because Child is a subclass of Parent
print(isinstance(g, Child))   # Output: True, because Grandchild is a subclass of Child
print(isinstance(g, Parent))  # Output: True, because Grandchild is a subclass of Parent

# Checking subclasses
print(issubclass(Child, Parent))       # Output: True
print(issubclass(Grandchild, Parent))  # Output: True
print(issubclass(Grandchild, Child))   # Output: True
print(issubclass(Parent, Child))       # Output: False

• isinstance(object, classinfo) returns True if object is an instance of classinfo or a subclass thereof.

• issubclass(class, classinfo) returns True if class is a subclass of classinfo.

These functions are handy for ensuring your objects and classes inherit from the expected parent classes. 

 

 isinstance(object, class):

• Checks if the object is an instance of the given class or its subclass.
• Useful when you have an object and want to verify its type.

issubclass(subclass, class):

• Checks if the given class is a subclass of another class.
• Useful when you’re dealing with class relationships and not specific instances.

 

If you want to check if an object belongs to a child class (and not the parent class directly):

if isinstance(c, Parent) and not isinstance(g, Child):
    print("Object belongs to ChildA but not Parent directly.")

False

 

Shallow copy vs deep copy in python

 

'''
Shallow Copy (copy.copy):

   A shallow copy creates a new object and then (to the extent possible) inserts references into it to the
   objects found in the original object. This means that if you modify a sub-object of the original object,
   the same modification will be reflected in the copied object.

    Creates a new object and copies the references of the original object's internal data.
    If the original object contains references to other objects, the shallow copy will share those same references.
    Changes made to the original object or the nested objects will also affect the shallow copy.
'''

import copy

original_list = [[1, 2], [3, 4]]

# Create a shallow copy
shallow_copy_list = copy.copy(original_list)

# Modify the original list
original_list.append([5, 6])
original_list[0][0] = 'X'

print("Original List:", original_list)
print("Shallow Copy List:", shallow_copy_list)

'''
output:
Original List: [['X', 2], [3, 4], [5, 6]]
Shallow Copy List: [['X', 2], [3, 4]]

As you can see, when we modified the original list by changing the first element of the first sublist, 
the same change was reflected in the shallow copy. 
However, when we appended a new sublist to the original list, the shallow copy remained unchanged.

2nd example:

A shallow copy creates a new object, but it "inserts references" into it to the objects found in the original. 

This means that if the original object contains references to other objects, 
these are shared between the original and the copied object.

You can create a shallow copy using the copy module's copy() method, the copy method of a list, or slicing.

import copy
original = [[1, 2, 3], [4, 5, 6]]
shallow_copy = copy.copy(original)
shallow_copy[0][0] = 'X'   -- Suppose we have modified the shallow copied object element , it will update in original 
print(original)  # Output: [['X', 2, 3], [4, 5, 6]]
print(shallow_copy)  # Output: [['X', 2, 3], [4, 5, 6]]

'''

'''
Deep Copy
A deep copy creates a new compound object and then, recursively, "inserts copies"
into it of the objects found in the original.
A deep copy creates a new object and recursively adds the copies of nested objects found in the original, 
meaning that it does not share references with the original object.
'''
import copy

original_list = [[1, 2], [3, 4]]

# Create a deep copy
deep_copy_list = copy.deepcopy(original_list)

# Modify the original list
original_list.append([5, 6])
original_list[0][0] = 'X'

print("Original List:", original_list)
print("Deep Copy List:", deep_copy_list)

''' output 
Original List: [['X', 2], [3, 4], [5, 6]]
Deep Copy List: [[1, 2], [3, 4]]

In this case, the deep copy remained completely unchanged, even when we modified the original list. 
This is because the deep copy created entirely new objects, rather than just referencing the original objects.


2nd example:

import copy
original = [[1, 2, 3], [4, 5, 6]]
deep_copy = copy.deepcopy(original)
deep_copy[0][0] = 'X' ---- Suppose we have modified the deep copied object element , it will not change the original obj.
print(original)  # Output: [[1, 2, 3], [4, 5, 6]]
print(deep_copy)  # Output: [['X', 2, 3], [4, 5, 6]]

use a shallow copy when you need to copy the object structure without copying the elements inside, and 
use a deep copy when you need to clone the object along with all the objects it references.

'''