# Iniciar el servidor de Kafka
bin/kafka-server-start.sh config/server.properties

# Crear un tópico para la ingesta de datos
bin/kafka-topics.sh --create --topic ingesta-datos --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# Crear una sesión de Spark
spark = SparkSession.builder.appName("LambdaArchitecture").getOrCreate()

# Definir el esquema de los datos
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("valor", StringType(), True)
])

# Leer datos en tiempo real desde Kafka
df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "ingesta-datos").load()

# Convertir los datos de Kafka a un DataFrame de Spark
df = df.selectExpr("CAST(value AS STRING)").select(from_json(col("value"), schema).alias("data")).select("data.*")

# Procesar los datos en tiempo real
query = df.writeStream.outputMode("append").format("console").start()

query.awaitTermination()

# Leer datos históricos desde una fuente de almacenamiento (por ejemplo, HDFS)
historical_df = spark.read.format("csv").option("header", "true").load("hdfs://path/to/historical/data")

# Procesar los datos históricos
historical_df.createOrReplaceTempView("historical_data")
result = spark.sql("SELECT id, COUNT(*) as count FROM historical_data GROUP BY id")

# Guardar los resultados procesados
result.write.format("parquet").save("hdfs://path/to/save/results")

# Iniciar el servidor de Kafka
bin/kafka-server-start.sh config/server.properties

# Crear un tópico para la ingesta de datos
bin/kafka-topics.sh --create --topic ingesta-datos --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import java.util.Properties;

public class KappaArchitecture {
    public static void main(String[] args) throws Exception {
        // Configurar el entorno de ejecución de Flink
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Configurar las propiedades de Kafka
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "flink-group");

        // Crear un consumidor de Kafka
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("ingesta-datos", new SimpleStringSchema(), properties);

        // Leer datos en tiempo real desde Kafka
        DataStream<String> stream = env.addSource(consumer);

        // Procesar los datos en tiempo real
        DataStream<String> processedStream = stream.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                // Procesamiento de datos (ejemplo simple)
                return "Procesado: " + value;
            }
        });

        // Imprimir los resultados procesados
        processedStream.print();

        // Ejecutar el entorno de Flink
        env.execute("Kappa Architecture");
    }
}

-- Crear una tabla en Cassandra
CREATE KEYSPACE kappa WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1};
CREATE TABLE kappa.resultados (id UUID PRIMARY KEY, valor TEXT);

-- Insertar datos procesados en Cassandra (desde Flink)
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Session;

public class CassandraConnector {
    private Cluster cluster;
    private Session session;

    public void connect(String node) {
        cluster = Cluster.builder().addContactPoint(node).build();
        session = cluster.connect();
    }

    public void close() {
        session.close();
        cluster.close();
    }

    public void insertData(String id, String valor) {
        session.execute("INSERT INTO kappa.resultados (id, valor) VALUES (" + id + ", '" + valor + "')");
    }
}