۱۱ نوع ساختار داده پایتون (Python Data Structures) برای تحلیل داده چیست؟

# Define a list
demo_list = [1, 2, 3, 4, 5]

# Append an element
demo_list.append(6)
print("After append:", demo_list)

# Insert at index
demo_list.insert(2, 8)
print("After insert:", demo_list)

# Extend with another list
demo_list.extend([9, 10, 11])
print("After extend:", demo_list)

# Index of an element
index = demo_list.index(4)
print("Index of 4:", index)

# Remove an element
demo_list.remove(3)
print("After remove:", demo_list)

# Define a dictionary
my_dict = {    
    "name": "Siya",    
    "age": 26,    
    "city": "New York"
}

# Python 3.9+ merge operators
additional_info = {"profession": "Data Scientist", "experience": 5}

complete_dict = my_dict | additional_info  # Merge operator
my_dict |= additional_info  # Update operator

print("Name:", my_dict["name"])
print("Age:", my_dict["age"])
print("City:", my_dict["city"])

my_set = {1, 2, 3, 4, 5}
my_set.add(6)
print("After adding 6:", my_set)

my_set.discard(3)
print("After removing 3:", my_set)

print("Is 2 in the set?", 2 in my_set)

new_set = {4, 5, 6, 7, 8}
print("Union:", my_set.union(new_set))
print("Intersection:", my_set.intersection(new_set))

my_tuple = (1, 2, 3, 4, 5, 6, 3, 3)

print("Count of 3:", my_tuple.count(3))
print("Index of 4:", my_tuple.index(4))

# Named tuples for better readability (Python 3.9+ supports improved type hints)
from typing import NamedTuple

class DataPoint(NamedTuple):
    timestamp: str
    value: float
    category: str

data = DataPoint("2024-01-01", 42.5, "sales")
print(f"Value: {data.value}, Category: {data.category}")

stack = []
stack.append('k')
stack.append('l')
stack.append('m')

print(stack)          # ['k', 'l', 'm']
print(stack.pop())    # 'm'
print(stack.pop())    # 'l'
print(stack.pop())    # 'k'

from collections import deque

stack = deque()
stack.append('x')
stack.append('y')
stack.append('z')

print(stack)          # deque(['x', 'y', 'z'])
print(stack.pop())    # 'z'
print(stack.pop())    # 'y'
print(stack.pop())    # 'x'

class Node:
    def __init__(self, data):
        self.data = data
        self.next = None

class LinkedList:
    def __init__(self):
        self.head = None

    # Insert at beginning
    def insertAtBeginning(self, new_data):
        new_node = Node(new_data)
        new_node.next = self.head
        self.head = new_node

    def display(self):
        elements = []
        current = self.head
        while current:
            elements.append(current.data)
            current = current.next
        return elements

queue = []
queue.append(5)
queue.append(7)
queue.append(9)

print(queue)          # [5, 7, 9]
print(queue.pop(0))   # 5
print(queue)          # [7, 9]

from collections import deque

queue = deque()
queue.append(1)
queue.append(2)
queue.append(3)

print(queue.popleft())  # 1 - O(1) operation

import heapq

heap = []
heapq.heappush(heap, 5)
heapq.heappush(heap, 3)
heapq.heappush(heap, 7)
heapq.heappush(heap, 1)

print(heapq.heappop(heap))       # 1
print(heapq.nlargest(2, heap))   # [7, 5]
print(heapq.nsmallest(2, heap))  # [3, 5]

import numpy as np

arr1 = np.array([1, 2, 3, 4])
print("NumPy array:", arr1)

arr2 = np.zeros(5)
print("Zeros:", arr2)

arr3 = np.arange(1, 5)
print("Range:", arr3)

# Vectorized operations (much faster than loops)
arr4 = np.array([1, 2, 3, 4])
arr5 = np.array([5, 6, 7, 8])

result = arr4 * arr5  # Element-wise multiplication
print("Vectorized multiplication:", result)

import pandas as pd

courses = pd.Series(["Hadoop", "Spark", "Python", "Oracle"])
print(courses[2])   # Python

# Series with custom index
sales_data = pd.Series([100, 150, 200], index=['Jan', 'Feb', 'Mar'])
print(sales_data['Feb'])  # 150

import pandas as pd

data = {
  "Score": [580, 250, 422],
  "id": [22, 37, 55],
  "category": ["A", "B", "A"]
}

df = pd.DataFrame(data)
print(df)

# Advanced DataFrame operations
grouped = df.groupby('category')['Score'].mean()
print("Average score by category:")
print(grouped)

from collections import Counter

colors = ['red', 'blue', 'red', 'green', 'blue', 'blue']
color_counts = Counter(colors)

print(color_counts)             # Counter({'blue': 3, 'red': 2, 'green': 1})
print(color_counts.most_common(2))  # [('blue', 3), ('red', 2)]

# Counter arithmetic operations
more_colors = Counter(['red', 'red', 'yellow'])
combined = color_counts + more_colors

print("Combined counts:", combined)

text = "  Data Analysis with Python  "
cleaned = text.strip()
words = cleaned.split()

print(cleaned.lower())                  # data analysis with python
print(cleaned.replace("Python", "R"))   # Data Analysis with R

# Advanced string operations
import re

email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z-0-9.-]+\.[A-Z|a-z]{2,}\b'
sample_text = "Contact us at info@company.com or sales@company.org"
emails = re.findall(email_pattern, sample_text)

print("Extracted emails:", emails)

import numpy as np

matrix = np.array([[1, 2, 3],
                   [۴, ۵, ۶],
                   [۷, ۸, ۹]])

print("Original matrix:")
print(matrix)

print("Transpose:")
print(matrix.transpose())

print("Matrix multiplication:")
print(matrix.dot(matrix))

# Statistical operations
print("Mean:", matrix.mean())
print("Standard deviation:", matrix.std())

# Efficient dictionary merging (Python 3.9+)
config_defaults = {"timeout": 30, "retries": 3}
user_config = {"timeout": 60, "debug": True}

# New approach - cleaner and efficient
final_config = config_defaults | user_config

# Bulk updates for data transformation
data_batch = [{"id": 1, "value": 100}, {"id": 2, "value": 200}]
lookup_cache = {}

for item in data_batch:
    lookup_cache |= {item["id"]: item["value"]}

# Pattern matching for complex data structures
def process_data_record(record):
    match record:
        case {"type": "user", "id": user_id, "data": user_data}:
            return process_user_data(user_id, user_data)
        case {"type": "transaction", "amount": amount, "currency": "USD"}:
            return process_usd_transaction(amount)
        case {"type": "error", "message": msg}:
            return handle_error(msg)
        case _:
            return handle_unknown_format(record)

# Memory-efficient data processing
def process_large_dataset(filename):
    # Generator expression - processes one line at a time
    cleaned_lines = (line.strip().lower() for line in open(filename)
                     if line.strip() and not line.startswith('#'))

    # Chained processing without intermediate storage
    parsed_data = (json.loads(line) for line in cleaned_lines)

    valid_records = (record for record in parsed_data
                     if 'required_field' in record)

    return valid_records


# Type-specific optimizations
import array

# Use array.array for homogeneous numeric data (50% less memory)
numeric_data = array.array('i', [1, 2, 3, 4, 5])  # 'i' for integers

import asyncio
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp

# Async processing for I/O-bound data operations
async def fetch_and_process_data(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_single_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    return results

# Process-based parallelism for CPU-intensive data transformations
def parallel_data_transform(data_chunks):
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(transform_chunk, data_chunks))
    return results

import asyncio
from collections import defaultdict, deque

async def stream_processor(data_source):
    # Use deque for efficient sliding window operations
    window = deque(maxlen=1000)

    # Dictionary for fast lookup during data enrichment
    lookup_cache = {}

    # Process streaming data with generators
    async for record in data_source:
        # Fast membership testing with sets
        if record['id'] not in processed_ids:
            enriched_record = enrich_data(record, lookup_cache)
            window.append(enriched_record)
            yield enriched_record


def enrich_data(record, cache):
    # Efficient dictionary-based enrichment
    return {**record, "enriched_field": cache.get(record['key'], 'default')}

import pandas as pd
from typing import Dict, List

def batch_etl_pipeline(source_config: Dict, transformations: List):
    # Use dictionaries for configuration management
    connection_params = {
        'host': source_config['database']['host'],
        'credentials': source_config['database']['credentials']
    }

    # Lists for transformation pipeline management
    applied_transformations = []

    # Load data into DataFrame for vectorized operations
    df = pd.read_sql(source_config['query'], connection_params)

    # Apply transformations using list iteration
    for transform_func in transformations:
        df = transform_func(df)
        applied_transformations.append(transform_func.__name__)

    # Dictionary for pipeline metadata
    pipeline_metadata = {
        'processed_records': len(df),
        'transformations_applied': applied_transformations,
        'processing_timestamp': pd.Timestamp.now()
    }

    return df, pipeline_metadata

from typing import Set, Dict, Any
import pandas as pd
from collections import defaultdict

class DataValidator:
    def __init__(self, schema_config: Dict):
        self.required_fields: Set[str] = set(schema_config['required'])
        self.field_types: Dict[str, type] = schema_config['types']
        self.allowed_values: Dict[str, Set] = {
            field: set(values) for field, values in schema_config.get('enums', {}).items()
        }

    def validate_batch(self, data: pd.DataFrame) -> Dict[str, Any]:
        validation_results = {
            'valid_records': 0,
            'errors': defaultdict(list),
            'field_stats': {}
        }

        # Check required fields using set operations
        missing_fields = self.required_fields - set(data.columns)
        if missing_fields:
            validation_results['errors']['missing_fields'] = list(missing_fields)

        # Validate allowed values using set membership
        for field, allowed in self.allowed_values.items():
            if field in data.columns:
                invalid_values = set(data[field].dropna()) - allowed
                if invalid_values:
                    validation_results['errors'][f'invalid_{field}'] = list(invalid_values)

        return validation_results

import numpy as np
from collections import Counter, defaultdict

def optimize_data_structures(data_characteristics):
    """
    Select optimal data structures based on data characteristics
    """
    recommendations = {}

    # For frequent lookups: dictionary vs set vs list
    if data_characteristics['lookup_heavy']:
        if data_characteristics['key_value_pairs']:
            recommendations['primary'] = 'dictionary'
        else:
            recommendations['primary'] = 'set'

    # For numerical operations: NumPy arrays
    if data_characteristics['numerical_computation']:
        recommendations['numerical'] = 'numpy_array'

    # For counting operations: Counter
    if data_characteristics['frequency_analysis']:
        recommendations['counting'] = 'collections.Counter'

    # For queue operations: deque vs list
    if data_characteristics['queue_operations']:
        recommendations['queue'] = 'collections.deque'

    return recommendations


# Example usage for different scenarios
def process_by_scenario(data, scenario_type):
    if scenario_type == 'real_time_analytics':
        # Use deque for sliding windows, dict for fast lookups
        window = deque(maxlen=10000)
        lookup_index = {}

    elif scenario_type == 'batch_aggregation':
        # Use Counter for frequency analysis, defaultdict for grouping
        frequency_counter = Counter()
        grouped_data = defaultdict(list)

    elif scenario_type == 'numerical_analysis':
        # Use NumPy arrays for vectorized operations
        numeric_array = np.array(data)
        return np.mean(numeric_array), np.std(numeric_array)

# Integration with Apache Arrow for efficient data exchange
import pyarrow as pa
import pandas as pd

def arrow_integration_example(pandas_df):
    # Convert DataFrame to Arrow Table for efficient serialization
    arrow_table = pa.Table.from_pandas(pandas_df)
    # Zero-copy conversion back to pandas when needed
    converted_df = arrow_table.to_pandas()
    # Efficient columnar operations
    column_data = arrow_table['column_name'].to_pylist()
    return column_data


# Integration with distributed processing frameworks
def spark_integration_pattern(spark_session, data):
    # Convert Python data structures to Spark DataFrames
    schema_dict = {'id': 'int', 'value': 'double', 'category': 'string'}
    spark_df = spark_session.createDataFrame(data, schema=schema_dict)
    # Process using Spark's distributed capabilities
    result = spark_df.groupBy('category').agg({'value': 'avg'})
    # Convert back to Python data structures for further processing
    python_result = [row.asDict() for row in result.collect()]
    return python_result