Kafka

ALUMNOS/MDAA/ANTONI_MOLLA/KAFKA/src/main/python/.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "python.analysis.autoImportCompletions": true
+}

ALUMNOS/MDAA/ANTONI_MOLLA/KAFKA/src/main/python/entregable/barcelona.txt

@@ -0,0 +1,16 @@
+{"nombre": "Marc-André ter Stegen", "posición": "Portero", "edad": 32, "nacionalidad": "Alemania", "partidos_jugados": 23, "goles": 0, "asistencias": 0, "minutos_jugados": 2070, "tarjetas_amarillas": 1, "tarjetas_rojas": 0, "valor_mercado": 15.00},
+{"nombre": "Iñaki Peña", "posición": "Portero", "edad": 25, "nacionalidad": "España", "partidos_jugados": 5, "goles": 0, "asistencias": 0, "minutos_jugados": 450, "tarjetas_amarillas": 0, "tarjetas_rojas": 0, "valor_mercado": 3.00},
+{"nombre": "Ronald Araújo", "posición": "Defensa", "edad": 26, "nacionalidad": "Uruguay", "partidos_jugados": 22, "goles": 3, "asistencias": 1, "minutos_jugados": 1980, "tarjetas_amarillas": 4, "tarjetas_rojas": 0, "valor_mercado": 60.00},
+{"nombre": "Jules Koundé", "posición": "Defensa", "edad": 26, "nacionalidad": "Francia", "partidos_jugados": 20, "goles": 2, "asistencias": 2, "minutos_jugados": 1800, "tarjetas_amarillas": 3, "tarjetas_rojas": 0, "valor_mercado": 55.00},
+{"nombre": "Andreas Christensen", "posición": "Defensa", "edad": 28, "nacionalidad": "Dinamarca", "partidos_jugados": 18, "goles": 1, "asistencias": 0, "minutos_jugados": 1620, "tarjetas_amarillas": 2, "tarjetas_rojas": 0, "valor_mercado": 40.00},
+{"nombre": "Alejandro Balde", "posición": "Defensa", "edad": 21, "nacionalidad": "España", "partidos_jugados": 21, "goles": 1, "asistencias": 4, "minutos_jugados": 1890, "tarjetas_amarillas": 5, "tarjetas_rojas": 0, "valor_mercado": 50.00},
+{"nombre": "Sergi Roberto", "posición": "Centrocampista", "edad": 33, "nacionalidad": "España", "partidos_jugados": 15, "goles": 2, "asistencias": 3, "minutos_jugados": 1350, "tarjetas_amarillas": 2, "tarjetas_rojas": 0, "valor_mercado": 10.00},
+{"nombre": "Frenkie de Jong", "posición": "Centrocampista", "edad": 27, "nacionalidad": "Países Bajos", "partidos_jugados": 20, "goles": 3, "asistencias": 5, "minutos_jugados": 1800, "tarjetas_amarillas": 3, "tarjetas_rojas": 0, "valor_mercado": 75.00},
+{"nombre": "Gavi", "posición": "Centrocampista", "edad": 20, "nacionalidad": "España", "partidos_jugados": 22, "goles": 4, "asistencias": 6, "minutos_jugados": 1980, "tarjetas_amarillas": 6, "tarjetas_rojas": 1, "valor_mercado": 70.00},
+{"nombre": "Pedri", "posición": "Centrocampista", "edad": 22, "nacionalidad": "España", "partidos_jugados": 19, "goles": 5, "asistencias": 7, "minutos_jugados": 1710, "tarjetas_amarillas": 4, "tarjetas_rojas": 0, "valor_mercado": 80.00},
+{"nombre": "Ilkay Gündogan", "posición": "Centrocampista", "edad": 34, "nacionalidad": "Alemania", "partidos_jugados": 18, "goles": 2, "asistencias": 4, "minutos_jugados": 1620, "tarjetas_amarillas": 3, "tarjetas_rojas": 0, "valor_mercado": 25.00},
+{"nombre": "Robert Lewandowski", "posición": "Delantero", "edad": 36, "nacionalidad": "Polonia", "partidos_jugados": 20, "goles": 15, "asistencias": 3, "minutos_jugados": 1800, "tarjetas_amarillas": 2, "tarjetas_rojas": 0, "valor_mercado": 30.00},
+{"nombre": "Raphinha", "posición": "Delantero", "edad": 28, "nacionalidad": "Brasil", "partidos_jugados": 19, "goles": 7, "asistencias": 5, "minutos_jugados": 1710, "tarjetas_amarillas": 4, "tarjetas_rojas": 0, "valor_mercado": 50.00},
+{"nombre": "Ansu Fati", "posición": "Delantero", "edad": 22, "nacionalidad": "España", "partidos_jugados": 18, "goles": 6, "asistencias": 4, "minutos_jugados": 1620, "tarjetas_amarillas": 3, "tarjetas_rojas": 0, "valor_mercado": 40.00},
+{"nombre": "Lamine Yamal", "posición": "Delantero", "edad": 17, "nacionalidad": "España", "partidos_jugados": 15, "goles": 4, "asistencias": 2, "minutos_jugados": 1350, "tarjetas_amarillas": 2, "tarjetas_rojas": 0, "valor_mercado": 25.00},
+{"nombre": "Ferran Torres", "posición": "Delantero", "edad": 25, "nacionalidad": "España", "partidos_jugados": 17, "goles": 5, "asistencias": 3, "minutos_jugados": 1530, "tarjetas_amarillas": 3, "tarjetas_rojas": 0, "valor_mercado": 35.00}

ALUMNOS/MDAA/ANTONI_MOLLA/KAFKA/src/main/python/entregable/consumer.py

@@ -0,0 +1,47 @@
+#SIMPLE
+
+from confluent_kafka import Consumer, KafkaError
+import json
+
+# Configuración del consumidor
+config = {
+    'bootstrap.servers': 'localhost:9092',  
+    'group.id': 'python-consumer-group',
+    'auto.offset.reset': 'earliest'  
+}
+
+# Crear un consumidor
+consumer = Consumer(config)
+
+# Suscribirse a un tópico
+topic = 'barcelona'  # El nombre del tópico
+consumer.subscribe([topic])
+
+# Loop infinito de consumo de mensajes del topic
+try:
+    while True:
+        msg = consumer.poll(1.0)  # Lee nuevos mensajes cada 1 segundo
+
+
+        if msg is None:
+            continue
+        if msg.error():
+            if msg.error().code() == KafkaError._PARTITION_EOF:
+                print("No hay más mensajes en esta partición.")
+            else:
+                print("Error al recibir mensaje: {}".format(msg.error()))
+        else:
+            # Procesar el mensaje recibido. Aqui solo lo mostramos por pantall. En una App real se pueden hacer cualquier
+            # cosa con un mensaje. Ejmplos: Filtrarlo, modificarlo, guardarlo en una base de datos, 
+            #enviarlo a otra applicación, etc.
+            print('Nuevo jugador: {}'.format(msg.value().decode('utf-8')))
+
+except KeyboardInterrupt:
+    pass
+finally:
+    # Cerrar el consumidor al parar la Applicacion Python
+    consumer.close()
+
+
+
+

ALUMNOS/MDAA/ANTONI_MOLLA/KAFKA/src/main/python/entregable/producer.py

@@ -0,0 +1,33 @@
+import time
+from json import dumps
+from confluent_kafka import Producer
+import re
+
+# Configuración del productor
+config = {
+    'bootstrap.servers': 'localhost:9092',  # Cambia esto con la dirección de tu servidor Kafka
+    'client.id': 'python-producer'
+}
+
+# Crear un productor
+producer = Producer(config)
+
+
+topic_kafka = 'barcelona'  # El nombre del tópico
+
+file1 = open('barcelona.txt',encoding="utf8")
+Lines = file1.readlines()
+
+count = 0
+# Strips the newline character
+for line in Lines:
+    time.sleep(2)
+    print( line.strip() + "\n")
+    producer.produce(topic=topic_kafka, value=line, key='key1')  # Send bytes
+    # After your loop where you send messages:
+    producer.flush()
+
+
+# Optionally, you can check if there are any messages that failed to be delivered:
+if producer.flush() != 0:
+    print("Some messages failed to be delivered")