# Premier programme big-data
print("Bonjour, monde !")

# Variables
nom = "NiangProgrammeur"
age = 25

# F-strings pour formater les chaînes (big-data 3.6+)
print(f"Je m'appelle {nom} et j'ai {age} ans")

# Opérations simples
resultat = 10 + 5
print(f"10 + 5 = {resultat}")

# Définition d'une fonction
def calculer_moyenne(nombres):
    """Calcule la moyenne d'une liste de nombres."""
    if len(nombres) == 0:
        return 0
    somme = sum(nombres)
    moyenne = somme / len(nombres)
    return moyenne

# Utilisation
notes = [15, 18, 12, 20, 16]
moyenne = calculer_moyenne(notes)
print(f"La moyenne est : {moyenne}")

# Déclaration de variables
nom = "big-data"          # String (chaîne de caractères)
age = 30                # Integer (entier)
prix = 19.99            # Float (nombre décimal)
est_actif = True        # Boolean (booléen)
valeur_nulle = None     # NoneType (valeur nulle)

# Affichage
print(nom)
print(age)
print(prix)
print(est_actif)
print(valeur_nulle)

# Réassignation (changement de type)
variable = 10
print(type(variable))   # 

variable = "Dix"
print(type(variable))   # 

# Noms de variables valides
nom_utilisateur = "Bassirou"
age_utilisateur = 25
_privé = "variable privée"
CONSTANTE = 3.14159     # Convention pour les constantes

# Exemple conceptuel de MapReduce
# Phase Map : Traiter les données et générer des paires clé-valeur

def map_function(document):
    """Divise le document en mots et compte les occurrences"""
    words = document.split()
    result = []
    for word in words:
        result.append((word.lower(), 1))
    return result

# Phase Reduce : Agréger les valeurs par clé

def reduce_function(key, values):
    """Somme toutes les occurrences d'un mot"""
    return (key, sum(values))

# Exemple d'utilisation
documents = [
    "Le Big Data est important",
    "Le Big Data transforme les entreprises",
    "Les données sont importantes"
]

# Phase Map
mapped_data = []
for doc in documents:
    mapped_data.extend(map_function(doc))
# Résultat: [('le', 1), ('big', 1), ('data', 1), ('est', 1), ...]

# Phase Shuffle & Sort (groupement par clé)
# Simulé ici - en réalité fait automatiquement par Hadoop
grouped = {}
for key, value in mapped_data:
    if key not in grouped:
        grouped[key] = []
    grouped[key].append(value)

# Phase Reduce
reduced_data = []
for key, values in grouped.items():
    reduced_data.append(reduce_function(key, values))
# Résultat: [('le', 2), ('big', 2), ('data', 2), ('est', 1), ...]

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, sum, count

# Créer une session Spark
spark = SparkSession.builder \
    .appName("BigDataExample") \
    .getOrCreate()

# Lire des données depuis HDFS ou un fichier local
df = spark.read.csv("hdfs://namenode:9000/data/sales.csv", 
                    header=True, 
                    inferSchema=True)

# Opérations de transformation
# Filtrer les données
filtered_df = df.filter(col("amount") > 1000)

# Agrégations
summary = df.groupBy("region") \
    .agg(
        count("*").alias("total_sales"),
        sum("amount").alias("total_amount"),
        avg("amount").alias("avg_amount")
    )

# Jointures
df1 = spark.read.csv("hdfs://namenode:9000/data/customers.csv", 
                     header=True, inferSchema=True)
joined_df = df.join(df1, on="customer_id", how="inner")

# Écriture des résultats
summary.write.mode("overwrite") \
    .parquet("hdfs://namenode:9000/output/summary")

# Actions (déclenchent l'exécution)
results = summary.collect()
for row in results:
    print(f"Region: {row.region}, Total: {row.total_amount}")

spark.stop()

# Exemples de bases de données NoSQL pour Big Data

# MongoDB (Document Store)
from pymongo import MongoClient

client = MongoClient('mongodb://localhost:27017/')
db = client['bigdata_db']
collection = db['users']

# Insérer un document
user = {
    "name": "John Doe",
    "age": 30,
    "email": "john@example.com",
    "tags": ["developer", "data-scientist"]
}
collection.insert_one(user)

# Requête
result = collection.find({"age": {"$gt": 25}})

# Cassandra (Wide Column Store)
from cassandra.cluster import Cluster

cluster = Cluster(['127.0.0.1'])
session = cluster.connect('bigdata_keyspace')

# Créer une table
session.execute("""
    CREATE TABLE IF NOT EXISTS user_events (
        user_id UUID,
        event_time TIMESTAMP,
        event_type TEXT,
        data MAP,
        PRIMARY KEY (user_id, event_time)
    )
""")

# Insérer des données
session.execute("""
    INSERT INTO user_events (user_id, event_time, event_type, data)
    VALUES (uuid(), toTimestamp(now()), 'click', {'page': '/home', 'action': 'view'})
""")

# Redis (Key-Value Store)
import redis

r = redis.Redis(host='localhost', port=6379, db=0)

# Stocker des données
r.set('user:1000:name', 'John Doe')
r.set('user:1000:age', '30')

# Récupérer des données
name = r.get('user:1000:name')
age = r.get('user:1000:age')

# Traitement de flux avec Apache Kafka et Spark Streaming
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# Créer une session Spark avec support streaming
spark = SparkSession.builder \
    .appName("StreamProcessing") \
    .config("spark.sql.streaming.checkpointLocation", "/checkpoint") \
    .getOrCreate()

# Définir le schéma des données
schema = StructType([
    StructField("user_id", StringType(), True),
    StructField("event_type", StringType(), True),
    StructField("timestamp", StringType(), True),
    StructField("data", StringType(), True)
])

# Lire depuis Kafka
df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "user-events") \
    .load()

# Parser les données JSON
parsed_df = df.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select("data.*")

# Traitement en temps réel
processed_df = parsed_df \
    .filter(col("event_type") == "purchase") \
    .groupBy("user_id") \
    .count()

# Écrire les résultats (sink)
query = processed_df \
    .writeStream \
    .outputMode("update") \
    .format("console") \
    .start()

# Attendre la fin du traitement
query.awaitTermination()

# Data Warehouse vs Data Lake

# Data Warehouse (structuré, schéma défini)
# Exemple avec PostgreSQL/Greenplum
import psycopg2

conn = psycopg2.connect(
    host="datawarehouse.example.com",
    database="dw_production",
    user="analyst",
    password="password"
)
cursor = conn.cursor()

# Requête SQL structurée
cursor.execute("""
    SELECT 
        DATE_TRUNC('month', order_date) as month,
        region,
        SUM(amount) as total_sales
    FROM orders
    WHERE order_date >= '2024-01-01'
    GROUP BY month, region
    ORDER BY month, region
""")

# Data Lake (non structuré, schéma flexible)
# Exemple avec S3 + Spark
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("DataLakeAccess") \
    .getOrCreate()

# Lire depuis un Data Lake (S3, HDFS, etc.)
# Supporte JSON, Parquet, CSV, Avro, etc.
df_json = spark.read.json("s3://data-lake/raw/events/*.json")
df_parquet = spark.read.parquet("s3://data-lake/processed/analytics/")
df_csv = spark.read.csv("s3://data-lake/external/suppliers/*.csv", 
                       header=True, inferSchema=True)

# Union de différentes sources
combined_df = df_json.union(df_parquet).union(df_csv)

# Transformation et stockage
processed_df = combined_df \
    .filter(col("status") == "active") \
    .select("id", "name", "category", "price")

# Écrire dans le Data Lake (format Parquet pour performance)
processed_df.write.mode("overwrite") \
    .parquet("s3://data-lake/curated/products/")

# Fonction simple (sans paramètres)
def dire_bonjour():
    print("Bonjour !")

dire_bonjour()  # Appel de la fonction

# Fonction avec paramètres
def saluer(nom):
    return f"Bonjour, {nom} !"

message = saluer("big-data")
print(message)  # "Bonjour, big-data !"

# Fonction avec plusieurs paramètres
def additionner(a, b):
    return a + b

resultat = additionner(5, 3)
print(resultat)  # 8

# Fonction avec paramètres par défaut
def saluer_personne(nom, message="Bonjour"):
    return f"{message}, {nom} !"

print(saluer_personne("Bassirou"))              # "Bonjour, Bassirou !"
print(saluer_personne("Bassirou", "Salut"))     # "Salut, Bassirou !"

# Fonction avec arguments nommés
def creer_personne(nom, age, ville="Dakar"):
    return f"{nom}, {age} ans, habite à {ville}"

print(creer_personne("Bassirou", 25))
print(creer_personne(age=25, nom="Bassirou", ville="Thiès"))

# Fonction avec *args (arguments variables)
def additionner_nombres(*args):
    return sum(args)

print(additionner_nombres(1, 2, 3, 4, 5))  # 15

# Fonction avec **kwargs (arguments nommés variables)
def afficher_info(**kwargs):
    for cle, valeur in kwargs.items():
        print(f"{cle}: {valeur}")

afficher_info(nom="Bassirou", age=25, ville="Dakar")

# Fonction lambda (fonction anonyme)
carre = lambda x: x ** 2
print(carre(5))  # 25

# Utilisation de lambda avec map()
nombres = [1, 2, 3, 4, 5]
carres = list(map(lambda x: x ** 2, nombres))
print(carres)  # [1, 4, 9, 16, 25]

# Sécurité et Gouvernance Big Data

# Chiffrement des données au repos (HDFS)
# Configuration dans hdfs-site.xml
# dfs.encryption.key.provider.uri = kms://http@kms-host:9600/kms

# Chiffrement des données en transit (Kerberos)
from hdfs import InsecureClient
from hdfs.ext.kerberos import KerberosClient

# Client sécurisé avec Kerberos
client = KerberosClient('http://namenode:9870', principal='user@REALM')

# Contrôle d'accès basé sur les rôles (RBAC)
# Configuration dans Ranger ou Sentry
# Exemple de politique:
# {
#   "resource": {
#     "database": "sales_db",
#     "table": "transactions"
#   },
#   "permissions": {
#     "users": ["analyst", "data_scientist"],
#     "actions": ["SELECT", "INSERT"]
#   }
# }

# Audit et logging
import logging

logging.basicConfig(
    filename='bigdata_audit.log',
    level=logging.INFO,
    format='%(asctime)s - %(user)s - %(action)s - %(resource)s'
)

# Masquage des données sensibles
def mask_pii(data):
    """Masquer les informations personnelles identifiables"""
    masked = data.copy()
    masked['email'] = masked['email'].apply(lambda x: x[:3] + '***@***')
    masked['phone'] = masked['phone'].apply(lambda x: '***-***-' + x[-4:])
    return masked

# Gouvernance des données - Métadonnées
metadata = {
    'table_name': 'customer_data',
    'owner': 'data_team',
    'created_date': '2024-01-15',
    'last_updated': '2024-12-20',
    'schema_version': '1.2',
    'data_quality_score': 0.95,
    'sensitivity': 'high',
    'retention_policy': '7_years'
}

# Cas d'Usage Avancés du Big Data

# 1. Détection de fraudes en temps réel
from pyspark.sql.functions import window, count

# Analyser les transactions suspectes
fraud_patterns = transactions \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(
        window("timestamp", "5 minutes"),
        "user_id"
    ) \
    .agg(count("*").alias("transaction_count")) \
    .filter(col("transaction_count") > 10)  # Plus de 10 transactions en 5 min

# 2. Recommandation personnalisée
from pyspark.ml.recommendation import ALS

# Modèle de filtrage collaboratif
als = ALS(
    maxIter=10,
    regParam=0.1,
    userCol="user_id",
    itemCol="product_id",
    ratingCol="rating"
)
model = als.fit(ratings_df)

# Recommandations pour un utilisateur
user_recs = model.recommendForAllUsers(10)

# 3. Analyse de sentiment sur les réseaux sociaux
from pyspark.ml.feature import Tokenizer, HashingTF
from pyspark.ml.classification import NaiveBayes

# Traitement du texte
tokenizer = Tokenizer(inputCol="text", outputCol="words")
words_df = tokenizer.transform(social_media_posts)

# Extraction de caractéristiques
hashing_tf = HashingTF(inputCol="words", outputCol="features")
features_df = hashing_tf.transform(words_df)

# Classification de sentiment
nb = NaiveBayes(labelCol="sentiment", featuresCol="features")
sentiment_model = nb.fit(features_df)

# Carrières dans le Big Data

# 1. Data Engineer
# Responsabilités:
# - Concevoir et maintenir des pipelines de données
# - Optimiser les performances des systèmes Big Data
# - Gérer l'infrastructure Hadoop/Spark

# Compétences requises:
skills_data_engineer = [
    'Hadoop (HDFS, MapReduce)',
    'Apache Spark',
    'SQL et NoSQL',
    'Python/Java/Scala',
    'Kafka pour le streaming',
    'Cloud (AWS, Azure, GCP)'
]

# 2. Big Data Architect
# Responsabilités:
# - Concevoir l'architecture Big Data
# - Choisir les technologies appropriées
# - Planifier la scalabilité

# Compétences requises:
skills_architect = [
    'Architecture distribuée',
    'Design patterns Big Data',
    'Sécurité et gouvernance',
    'Performance tuning',
    'Cloud architecture'
]

# 3. Data Scientist (Big Data)
# Responsabilités:
# - Analyser de grands volumes de données
# - Développer des modèles ML à grande échelle
# - Extraire des insights business

# Compétences requises:
skills_data_scientist = [
    'Machine Learning distribué',
    'Spark MLlib',
    'Statistiques avancées',
    'Visualisation de données',
    'Domain expertise'
]

# 4. DevOps Engineer (Big Data)
# Responsabilités:
# - Automatiser les déploiements
# - Monitoring et alerting
# - CI/CD pour pipelines de données

# Compétences requises:
skills_devops = [
    'Docker, Kubernetes',
    'Terraform, Ansible',
    'Monitoring (Prometheus, Grafana)',
    'CI/CD (Jenkins, GitLab CI)',
    'Infrastructure as Code'
]

# Tendances Futures du Big Data

# 1. Edge Computing et IoT
# Traitement des données à la source (edge) avant envoi au cloud
edge_devices = [
    'Capteurs IoT',
    'Appareils mobiles',
    'Gateways intelligents',
    'Dispositifs embarqués'
]

# 2. Real-time Analytics
# Analyse en temps réel avec Apache Flink, Kafka Streams
from pyflink.datastream import StreamExecutionEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
stream = env.from_collection([1, 2, 3, 4, 5])
result = stream.map(lambda x: x * 2)
result.print()

# 3. Serverless Big Data
# Traitement sans serveur (AWS Lambda, Azure Functions)
# Avantages: Pas de gestion d'infrastructure, scaling automatique

# 4. Data Mesh Architecture
# Architecture décentralisée où chaque domaine gère ses propres données
data_mesh_principles = [
    'Domain-oriented decentralized data ownership',
    'Data as a product',
    'Self-serve data infrastructure',
    'Federated computational governance'
]

# 5. AI/ML Integration
# Intégration de l'IA dans les pipelines Big Data
from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import RandomForestRegressor

# Pipeline ML distribué
assembler = VectorAssembler(inputCols=["feature1", "feature2"], 
                           outputCol="features")
rf = RandomForestRegressor(featuresCol="features", labelCol="target")
pipeline = Pipeline(stages=[assembler, rf])
model = pipeline.fit(training_data)

# 6. Cloud-Native Big Data
# Services managés (AWS EMR, Azure HDInsight, GCP Dataproc)
# Avantages: Pas de gestion d'infrastructure, coûts optimisés

# 7. Data Governance et Privacy
# RGPD, CCPA compliance
# Techniques: Anonymisation, pseudonymisation, chiffrement