Introduction to LangChain

LangChain is revolutionizing how developers build AI-powered applications. It provides a framework for creating applications that leverage Large Language Models (LLMs) with advanced capabilities like memory, reasoning, and tool integration.

This comprehensive guide will take you from LangChain basics to building production-ready AI applications.

What is LangChain?

LangChain is a framework designed to simplify the creation of applications using large language models. It provides:


Chains: Sequences of calls to LLMs or other utilities
Agents: LLMs that make decisions about actions
Memory: Persistence between chain/agent calls
Indexes: Ways to structure documents for LLM interaction
Tools: Interfaces for LLMs to interact with external systems


Installation and Setup

Python Installation

pip install langchain langchain-openai langchain-community

# For specific integrations
pip install chromadb tiktoken python-dotenv

JavaScript/TypeScript Installation

npm install langchain @langchain/openai @langchain/community

# or
yarn add langchain @langchain/openai @langchain/community

Environment Setup

# .env file
OPENAI_API_KEY=your_openai_api_key_here
ANTHROPIC_API_KEY=your_anthropic_key_here

Basic LangChain Concepts

1. Simple LLM Call (Python)

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage

# Initialize the model
llm = ChatOpenAI(model="gpt-4", temperature=0.7)

# Create messages
messages = [
SystemMessage(content="You are a helpful AI assistant specialized in Python programming."),
HumanMessage(content="Explain list comprehensions with an example.")
]

# Get response
response = llm.invoke(messages)
print(response.content)

2. Prompt Templates

from langchain.prompts import ChatPromptTemplate, PromptTemplate

# Simple template
template = PromptTemplate(
input_variables=["language", "topic"],
template="Write a {language} function that {topic}"
)

prompt = template.format(language="Python", topic="sorts a list of integers")
print(prompt)

# Chat template
chat_template = ChatPromptTemplate.from_messages([
("system", "You are an expert {role}."),
("user", "Explain {concept} in simple terms.")
])

messages = chat_template.format_messages(
role="data scientist",
concept="gradient descent"
)

3. Chains: Combining Components

from langchain.chains import LLMChain
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0.7)

# Create a chain
chain = LLMChain(llm=llm, prompt=template)

# Run the chain
result = chain.run(language="JavaScript", topic="validates email addresses")
print(result)

Advanced LangChain Features

Memory: Making Conversations Contextual

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# Initialize memory
memory = ConversationBufferMemory()

# Create conversation chain
conversation = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)

# Have a conversation
response1 = conversation.predict(input="Hi, my name is Alice")
print(response1)

response2 = conversation.predict(input="What's my name?")
print(response2) # Will remember Alice

# Different memory types
from langchain.memory import ConversationSummaryMemory, ConversationBufferWindowMemory

# Summary memory - keeps a running summary
summary_memory = ConversationSummaryMemory(llm=llm)

# Window memory - keeps last N messages
window_memory = ConversationBufferWindowMemory(k=5)

Document Loading and Processing

from langchain.document_loaders import TextLoader, PyPDFLoader, WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Load documents
loader = PyPDFLoader("document.pdf")
documents = loader.load()

# Split into chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len
)

texts = text_splitter.split_documents(documents)
print(f"Split into {len(texts)} chunks")

Vector Stores and Embeddings

from langchain_openai import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA

# Create embeddings
embeddings = OpenAIEmbeddings()

# Create vector store
vectorstore = Chroma.from_documents(
documents=texts,
embedding=embeddings,
persist_directory="./chroma_db"
)

# Create retrieval chain
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)

# Ask questions about your documents
query = "What are the main points discussed in the document?"
result = qa_chain.run(query)
print(result)

Building an AI Agent

from langchain.agents import load_tools, initialize_agent, AgentType
from langchain_openai import ChatOpenAI

# Initialize LLM
llm = ChatOpenAI(temperature=0)

# Load tools
tools = load_tools(["serpapi", "llm-math"], llm=llm)

# Initialize agent
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)

# Run agent
response = agent.run(
"What was the average temperature in New York City last week? "
"Calculate the difference from the historical average."
)
print(response)

Custom Tools

from langchain.tools import Tool, tool
from langchain.agents import AgentExecutor

@tool
def search_database(query: str) -> str:
"""Search the company database for information."""
# Your database search logic here
return f"Results for: {query}"

@tool
def calculate_metrics(data: str) -> str:
"""Calculate business metrics from data."""
# Your calculation logic here
return "Calculated metrics: ..."

# Create agent with custom tools
tools = [search_database, calculate_metrics]

agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)

Production-Ready RAG Application

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

class RAGApplication:
def __init__(self, persist_directory="./vectorstore"):
self.llm = ChatOpenAI(model="gpt-4", temperature=0)
self.embeddings = OpenAIEmbeddings()
self.vectorstore = None
self.chain = None
self.persist_directory = persist_directory

def load_documents(self, documents):
"""Load and index documents."""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
texts = text_splitter.split_documents(documents)

self.vectorstore = Chroma.from_documents(
documents=texts,
embedding=self.embeddings,
persist_directory=self.persist_directory
)

# Create conversational chain
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)

self.chain = ConversationalRetrievalChain.from_llm(
llm=self.llm,
retriever=self.vectorstore.as_retriever(search_kwargs={"k": 5}),
memory=memory,
verbose=True
)

def query(self, question):
"""Query the RAG system."""
if not self.chain:
raise ValueError("Please load documents first")

result = self.chain({"question": question})
return result["answer"]

# Usage
app = RAGApplication()
app.load_documents(documents)
answer = app.query("What is the main topic of these documents?")

LangChain with Streaming

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain_openai import ChatOpenAI

# Initialize with streaming
llm = ChatOpenAI(
streaming=True,
callbacks=[StreamingStdOutCallbackHandler()],
temperature=0.7
)

# Stream response
response = llm.invoke("Write a short story about a robot learning to cook")

# Custom streaming callback
from langchain.callbacks.base import BaseCallbackHandler

class CustomStreamingCallback(BaseCallbackHandler):
def on_llm_new_token(self, token: str, **kwargs) -> None:
print(f"Token: {token}", end="", flush=True)

llm_with_custom_callback = ChatOpenAI(
streaming=True,
callbacks=[CustomStreamingCallback()]
)

Error Handling and Best Practices

from langchain.callbacks import get_openai_callback
import logging

# Track token usage
with get_openai_callback() as cb:
result = chain.run(query)
print(f"Total Tokens: {cb.total_tokens}")
print(f"Prompt Tokens: {cb.prompt_tokens}")
print(f"Completion Tokens: {cb.completion_tokens}")
print(f"Total Cost (USD): ${cb.total_cost}")

# Error handling
try:
result = chain.run(query)
except Exception as e:
logging.error(f"Error in LangChain execution: {e}")
# Implement fallback logic

Testing LangChain Applications

import pytest
from unittest.mock import Mock, patch

def test_chain_execution():
# Mock LLM responses
mock_llm = Mock()
mock_llm.invoke.return_value = "Mocked response"

chain = LLMChain(llm=mock_llm, prompt=template)
result = chain.run(language="Python", topic="test")

assert result == "Mocked response"
mock_llm.invoke.assert_called_once()

# Integration testing with actual API (use carefully)
@pytest.mark.integration
def test_real_llm_call():
llm = ChatOpenAI(model="gpt-3.5-turbo")
response = llm.invoke("Say 'test passed'")
assert "test passed" in response.content.lower()

Performance Optimization

1. Caching

from langchain.cache import InMemoryCache, SQLiteCache
import langchain

# In-memory cache
langchain.llm_cache = InMemoryCache()

# Persistent cache
langchain.llm_cache = SQLiteCache(database_path=".langchain.db")

2. Batch Processing

# Batch API calls for efficiency
inputs = [
{"language": "Python", "topic": "sorting"},
{"language": "JavaScript", "topic": "async"},
{"language": "Go", "topic": "concurrency"}
]

results = chain.batch(inputs)

Deployment Considerations


Environment Variables: Use secure secret management
Rate Limiting: Implement backoff strategies
Monitoring: Track token usage and costs
Error Recovery: Implement retry logic with exponential backoff
Vector Store: Use production-ready databases (Pinecone, Weaviate)


Conclusion

LangChain empowers developers to build sophisticated AI applications with ease. From simple chatbots to complex RAG systems and autonomous agents, LangChain provides the tools you need. Start with basic chains, experiment with agents, and scale to production-ready applications.

The AI development landscape is evolving rapidly—LangChain keeps you at the forefront.

What is ETL?

ETL (Extract, Transform, Load) is the backbone of modern data engineering. It's the process of collecting data from various sources, transforming it into a usable format, and loading it into a destination system for analysis and business intelligence.

In this comprehensive guide, we'll cover everything from ETL fundamentals to building production-ready data pipelines.

The Three Pillars of ETL

1. Extract
Gathering data from diverse sources:

Databases (SQL, NoSQL)
APIs (REST, GraphQL)
Files (CSV, JSON, Parquet)
Streaming data (Kafka, Kinesis)
Web scraping
Cloud storage (S3, Azure Blob)


2. Transform
Converting data into a usable format:

Data cleaning and validation
Type conversion and formatting
Filtering and aggregation
Joining multiple sources
Calculating derived metrics
Data enrichment


3. Load
Writing data to target systems:

Data warehouses (Snowflake, BigQuery, Redshift)
Data lakes (S3, HDFS)
Databases
Analytics platforms


ETL vs ELT: Understanding the Difference



Aspect
ETL
ELT


Process Order
Transform before loading
Load then transform


Best For
Structured data, complex transformations
Big data, cloud-native architectures


Performance
Can be slower for large datasets
Leverages target system's power


Cost
Separate transformation servers
Uses destination compute



Building Your First ETL Pipeline with Python

Setup and Dependencies

pip install pandas sqlalchemy requests psycopg2-binary python-dotenv

Simple ETL Example

import pandas as pd
import requests
from sqlalchemy import create_engine
from datetime import datetime
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SimpleETL:
def __init__(self, db_connection_string):
self.engine = create_engine(db_connection_string)

def extract_from_api(self, api_url):
"""Extract data from REST API"""
logger.info(f"Extracting data from {api_url}")
try:
response = requests.get(api_url, timeout=30)
response.raise_for_status()
data = response.json()
return pd.DataFrame(data)
except Exception as e:
logger.error(f"Extraction failed: {e}")
raise

def extract_from_csv(self, file_path):
"""Extract data from CSV file"""
logger.info(f"Extracting data from {file_path}")
return pd.read_csv(file_path)

def extract_from_database(self, query):
"""Extract data from database"""
logger.info("Extracting data from database")
return pd.read_sql(query, self.engine)

def transform(self, df):
"""Transform the data"""
logger.info("Transforming data")

# Remove duplicates
df = df.drop_duplicates()

# Handle missing values
df = df.fillna({
'name': 'Unknown',
'age': df['age'].median(),
'email': ''
})

# Data type conversion
df['created_at'] = pd.to_datetime(df['created_at'])
df['age'] = df['age'].astype(int)

# Add derived columns
df['processed_at'] = datetime.now()
df['full_name'] = df['first_name'] + ' ' + df['last_name']

# Filter invalid records
df = df[df['age'] > 0]
df = df[df['email'].str.contains('@', na=False)]

# Standardize text
df['email'] = df['email'].str.lower().str.strip()

logger.info(f"Transformation complete. Records: {len(df)}")
return df

def load_to_database(self, df, table_name, if_exists='append'):
"""Load data to database"""
logger.info(f"Loading data to {table_name}")
try:
df.to_sql(
table_name,
self.engine,
if_exists=if_exists,
index=False,
method='multi',
chunksize=1000
)
logger.info(f"Successfully loaded {len(df)} records")
except Exception as e:
logger.error(f"Load failed: {e}")
raise

def run_pipeline(self, source_type, source, target_table):
"""Run the complete ETL pipeline"""
logger.info("Starting ETL pipeline")

try:
# Extract
if source_type == 'api':
df = self.extract_from_api(source)
elif source_type == 'csv':
df = self.extract_from_csv(source)
elif source_type == 'database':
df = self.extract_from_database(source)
else:
raise ValueError(f"Unknown source type: {source_type}")

# Transform
df_transformed = self.transform(df)

# Load
self.load_to_database(df_transformed, target_table)

logger.info("ETL pipeline completed successfully")
return True

except Exception as e:
logger.error(f"Pipeline failed: {e}")
return False

# Usage
if __name__ == "__main__":
db_url = "postgresql://user:password@localhost:5432/datawarehouse"
etl = SimpleETL(db_url)

# Run pipeline
etl.run_pipeline(
source_type='api',
source='https://api.example.com/users',
target_table='users_staging'
)

Advanced ETL with Apache Airflow

Installing Airflow

pip install apache-airflow

# Initialize database
airflow db init

# Create admin user
airflow users create
--username admin
--firstname Admin
--lastname User
--role Admin
--email admin@example.com

Airflow DAG Example

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime, timedelta
import pandas as pd

default_args = {
'owner': 'data_team',
'depends_on_past': False,
'start_date': datetime(2025, 1, 1),
'email': ['alerts@example.com'],
'email_on_failure': True,
'email_on_retry': False,
'retries': 3,
'retry_delay': timedelta(minutes=5),
}

dag = DAG(
'user_data_etl',
default_args=default_args,
description='ETL pipeline for user data',
schedule_interval='0 2 * * *', # Daily at 2 AM
catchup=False,
tags=['etl', 'users']
)

def extract_data(**context):
"""Extract data from source"""
# Your extraction logic
data = fetch_from_source()
# Push to XCom
context['ti'].xcom_push(key='raw_data', value=data)

def transform_data(**context):
"""Transform extracted data"""
# Pull from XCom
raw_data = context['ti'].xcom_pull(key='raw_data')

df = pd.DataFrame(raw_data)
# Transformation logic
df_transformed = apply_transformations(df)

context['ti'].xcom_push(key='transformed_data', value=df_transformed.to_dict())

def load_data(**context):
"""Load data to destination"""
transformed_data = context['ti'].xcom_pull(key='transformed_data')
df = pd.DataFrame(transformed_data)
# Loading logic
load_to_warehouse(df)

def validate_data(**context):
"""Validate loaded data"""
# Data quality checks
checks_passed = run_data_quality_checks()
if not checks_passed:
raise ValueError("Data quality checks failed")

# Define tasks
extract_task = PythonOperator(
task_id='extract_data',
python_callable=extract_data,
dag=dag
)

transform_task = PythonOperator(
task_id='transform_data',
python_callable=transform_data,
dag=dag
)

load_task = PythonOperator(
task_id='load_data',
python_callable=load_data,
dag=dag
)

validate_task = PythonOperator(
task_id='validate_data',
python_callable=validate_data,
dag=dag
)

# Set dependencies
extract_task >> transform_task >> load_task >> validate_task

Modern ETL with dbt (Data Build Tool)

dbt Model Example

-- models/staging/stg_users.sql
{{
config(
materialized='view'
)
}}

SELECT
user_id,
LOWER(TRIM(email)) as email,
CONCAT(first_name, ' ', last_name) as full_name,
CAST(created_at AS TIMESTAMP) as created_at,
CASE
WHEN age < 18 THEN 'minor'
WHEN age BETWEEN 18 AND 65 THEN 'adult'
ELSE 'senior'
END as age_group
FROM {{ source('raw', 'users') }}
WHERE email IS NOT NULL
AND created_at >= '2020-01-01'

-- models/marts/fct_user_activity.sql
{{
config(
materialized='table',
unique_key='user_id'
)
}}

WITH user_stats AS (
SELECT
user_id,
COUNT(*) as total_actions,
MIN(action_timestamp) as first_action,
MAX(action_timestamp) as last_action
FROM {{ ref('stg_user_actions') }}
GROUP BY user_id
)

SELECT
u.user_id,
u.email,
u.full_name,
us.total_actions,
us.first_action,
us.last_action,
DATEDIFF(day, us.first_action, us.last_action) as days_active
FROM {{ ref('stg_users') }} u
LEFT JOIN user_stats us ON u.user_id = us.user_id

Data Quality and Validation

import great_expectations as ge

def validate_data_quality(df):
"""Implement data quality checks"""

# Convert to Great Expectations DataFrame
ge_df = ge.from_pandas(df)

# Define expectations
ge_df.expect_column_to_exist('email')
ge_df.expect_column_values_to_not_be_null('user_id')
ge_df.expect_column_values_to_be_unique('user_id')
ge_df.expect_column_values_to_match_regex(
'email',
r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
)
ge_df.expect_column_values_to_be_between('age', 0, 120)

# Get validation results
results = ge_df.validate()

if not results['success']:
failed_expectations = [
exp for exp in results['results']
if not exp['success']
]
raise ValueError(f"Data quality checks failed: {failed_expectations}")

return True

Error Handling and Monitoring

import time
from functools import wraps

def retry_on_failure(max_retries=3, delay=5):
"""Decorator for retrying failed operations"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise
logger.warning(
f"Attempt {attempt + 1} failed: {e}. "
f"Retrying in {delay} seconds..."
)
time.sleep(delay)
return wrapper
return decorator

class ETLMonitor:
"""Monitor ETL pipeline performance"""

def __init__(self):
self.metrics = {
'start_time': None,
'end_time': None,
'records_extracted': 0,
'records_transformed': 0,
'records_loaded': 0,
'errors': []
}

def start(self):
self.metrics['start_time'] = datetime.now()

def end(self):
self.metrics['end_time'] = datetime.now()
self.metrics['duration'] = (
self.metrics['end_time'] - self.metrics['start_time']
).total_seconds()

def log_error(self, error):
self.metrics['errors'].append(str(error))

def report(self):
return {
'duration_seconds': self.metrics['duration'],
'records_processed': self.metrics['records_loaded'],
'success_rate': (
self.metrics['records_loaded'] /
self.metrics['records_extracted'] * 100
) if self.metrics['records_extracted'] > 0 else 0,
'errors': self.metrics['errors']
}

Performance Optimization

1. Parallel Processing

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import multiprocessing

def process_chunk(chunk):
"""Process a data chunk"""
# Transform logic here
return transformed_chunk

def parallel_transform(df, chunk_size=10000):
"""Transform data in parallel"""
chunks = [df[i:i+chunk_size] for i in range(0, len(df), chunk_size)]

with ProcessPoolExecutor(max_workers=multiprocessing.cpu_count()) as executor:
results = list(executor.map(process_chunk, chunks))

return pd.concat(results, ignore_index=True)

2. Incremental Loading

def incremental_extract(last_run_timestamp):
"""Extract only new/updated records"""
query = f"""
SELECT *
FROM source_table
WHERE updated_at > '{last_run_timestamp}'
ORDER BY updated_at
"""
return pd.read_sql(query, engine)

3. Batch Processing

def batch_load(df, batch_size=1000):
"""Load data in batches"""
for i in range(0, len(df), batch_size):
batch = df[i:i+batch_size]
load_batch_to_db(batch)
logger.info(f"Loaded batch {i//batch_size + 1}")

Cloud ETL Solutions

AWS Glue Example

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Extract
datasource = glueContext.create_dynamic_frame.from_catalog(
database="my_database",
table_name="source_table"
)

# Transform
transformed = ApplyMapping.apply(
frame=datasource,
mappings=[
("user_id", "string", "user_id", "string"),
("email", "string", "email", "string"),
("created_at", "string", "created_at", "timestamp")
]
)

# Load
glueContext.write_dynamic_frame.from_catalog(
frame=transformed,
database="my_database",
table_name="target_table"
)

job.commit()

Best Practices


Idempotency: Ensure pipelines can be safely re-run
Data Validation: Implement quality checks at each stage
Error Handling: Log errors and implement retry logic
Monitoring: Track pipeline metrics and set up alerts
Documentation: Document data lineage and transformations
Testing: Unit test transformations and integration test pipelines
Version Control: Keep ETL code in Git
Incremental Loading: Process only changed data when possible


Conclusion

ETL is the foundation of data-driven organizations. Whether you're building simple Python scripts or complex Airflow DAGs, the principles remain the same: extract reliably, transform accurately, and load efficiently.

Start small, monitor everything, and scale as your data needs grow. The journey from raw data to actionable insights starts with a solid ETL pipeline.

Introduction to Arrays

Arrays are the foundation of computer programming and one of the most fundamental data structures. Understanding arrays deeply is crucial for writing efficient code and solving complex algorithmic problems.

This comprehensive guide covers everything from basic array operations to advanced techniques used in competitive programming and technical interviews.

What Are Arrays?

An array is a contiguous block of memory that stores elements of the same data type. Each element can be accessed directly using its index, making arrays one of the most efficient data structures for random access.

Key Characteristics

Fixed Size: In most languages, arrays have a fixed size at creation
Contiguous Memory: Elements are stored sequentially in memory
O(1) Access: Direct access to any element by index
Same Type: All elements must be of the same data type


Array Operations and Time Complexity



Operation
Time Complexity
Description


Access
O(1)
Direct indexing


Search
O(n)
Linear scan (unsorted)


Insert (end)
O(1)*
*Amortized for dynamic arrays


Insert (middle)
O(n)
Requires shifting elements


Delete
O(n)
Requires shifting elements



Arrays in Different Languages

Python

# Lists in Python (dynamic arrays)
numbers = [1, 2, 3, 4, 5]

# Access
print(numbers[0]) # 1

# Slicing
print(numbers[1:4]) # [2, 3, 4]

# Common operations
numbers.append(6) # Add to end
numbers.insert(0, 0) # Insert at position
numbers.pop() # Remove last
numbers.remove(3) # Remove first occurrence of value

# List comprehension
squares = [x**2 for x in range(10)]

# Array module for fixed-type arrays
from array import array
int_array = array('i', [1, 2, 3, 4, 5])

JavaScript

// Arrays in JavaScript
const numbers = [1, 2, 3, 4, 5];

// Access
console.log(numbers[0]); // 1

// Common operations
numbers.push(6); // Add to end
numbers.unshift(0); // Add to beginning
numbers.pop(); // Remove last
numbers.shift(); // Remove first

// Functional methods
const doubled = numbers.map(x => x * 2);
const evens = numbers.filter(x => x % 2 === 0);
const sum = numbers.reduce((acc, x) => acc + x, 0);

// Spread operator
const copy = [...numbers];
const merged = [...numbers, ...otherArray];

Java

// Static array
int[] numbers = new int[5];
numbers[0] = 1;

// Array initialization
int[] nums = {1, 2, 3, 4, 5};

// ArrayList (dynamic)
ArrayList list = new ArrayList();
list.add(1);
list.add(2);
list.get(0);
list.remove(0);
list.size();

// Array to ArrayList
List arrayList = Arrays.asList(1, 2, 3, 4, 5);

// Stream operations
int sum = Arrays.stream(nums).sum();
int[] doubled = Arrays.stream(nums).map(x -> x * 2).toArray();

Essential Array Algorithms

1. Two Pointer Technique

def reverse_array(arr):
"""Reverse array in-place using two pointers"""
left, right = 0, len(arr) - 1

while left < right:
arr[left], arr[right] = arr[right], arr[left]
left += 1
right -= 1

return arr

def remove_duplicates_sorted(arr):
"""Remove duplicates from sorted array"""
if not arr:
return 0

write_idx = 1

for read_idx in range(1, len(arr)):
if arr[read_idx] != arr[read_idx - 1]:
arr[write_idx] = arr[read_idx]
write_idx += 1

return write_idx # New length

2. Sliding Window

def max_sum_subarray(arr, k):
"""Find maximum sum of k consecutive elements"""
if len(arr) < k:
return None

# Calculate sum of first window
window_sum = sum(arr[:k])
max_sum = window_sum

# Slide the window
for i in range(k, len(arr)):
window_sum = window_sum - arr[i - k] + arr[i]
max_sum = max(max_sum, window_sum)

return max_sum

def longest_substring_k_distinct(s, k):
"""Find longest substring with at most k distinct characters"""
char_count = {}
left = 0
max_length = 0

for right in range(len(s)):
char_count[s[right]] = char_count.get(s[right], 0) + 1

while len(char_count) > k:
char_count[s[left]] -= 1
if char_count[s[left]] == 0:
del char_count[s[left]]
left += 1

max_length = max(max_length, right - left + 1)

return max_length

3. Kadane's Algorithm (Maximum Subarray)

def max_subarray_sum(arr):
"""Find maximum sum of contiguous subarray"""
max_current = max_global = arr[0]

for i in range(1, len(arr)):
max_current = max(arr[i], max_current + arr[i])
max_global = max(max_global, max_current)

return max_global

# With indices
def max_subarray_with_indices(arr):
"""Return max sum and indices of subarray"""
max_current = arr[0]
max_global = arr[0]
start = end = s = 0

for i in range(1, len(arr)):
if arr[i] > max_current + arr[i]:
max_current = arr[i]
s = i
else:
max_current = max_current + arr[i]

if max_current > max_global:
max_global = max_current
start = s
end = i

return max_global, start, end

4. Binary Search

def binary_search(arr, target):
"""Binary search in sorted array"""
left, right = 0, len(arr) - 1

while left