<jupyter_start><jupyter_text>Cliente de la rama BATCHEste cliente se limita a leer los datos del topic y los vuelca al fichero historic.csv. Podríamos tener varios procesos iguales para permitir alta disponibilidad y mejor rendimiento, pero:- deberíamos asegurarnos de que varios procesos pueden escribir a la vez en el fichero,- en este caso particular (que no deja de ser un ejemplo muy sintético) incluso aunque el topic esté formado por varias particiones para facilitar el trabajo de la rama real-time, este cliente tan ligero que probablemente añadir varios no mejore el rendimiento.Si el destino fuera una base de datos probablemente podríamos obviar la primera consideración, aunque en el mundo streaming es más habitual volcar los datos a HDFS o S3. Recordemos que la escritura en HDFS en procesos Hadoop ocurre independientemente en cada tarea _reducer_ para conseguir paralelizar. Para paralelizar en streaming podemos buscar un efecto parecido: si tenemos varias particiones con un cliente por partición, la escritura de los clientes puede ser independiente si escriben en HDFS o en S3. En este caso perdemos el orden de los datos: los datos escritos en un mismo fichero por un mismo cliente sí pueden mantener el orden, pero de un fichero a otro no es posible. Una manera de solventar esto es escribir los ficheros en carpetas separadas por bloques de horas, además de que cada registro esté identificado por un timestamp.<jupyter_code>import pandas as pd
from ejercicios.houses import BATCH_GROUP, HISTORIC, TOPIC_SALES

ds = pd.read_csv(HISTORIC)
ds.head()
from confluent_kafka import Consumer, KafkaError
import json

c = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': BATCH_GROUP,
    'auto.offset.reset': 'earliest'
})

c.subscribe([TOPIC_SALES])

while True:
    msg = c.poll(1.0)

    if msg is None:
        continue
    if msg.error():
        print("Consumer error: {}".format(msg.error()))
        continue

    data_json = msg.value().decode('utf-8')
    data_dict = json.loads(data_json)

    print('Received message: {}:{}'.format(data_dict.get('price'), data_dict.get('yr_built')))

    ds = ds.append(data_dict, ignore_index=True)
    ds.to_csv(HISTORIC)

c.close()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.06586559857587895, 'fraction_numerical': 0.004005340453938585, 'mean_word_length': 4.592039800995025, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '460197', 'license': 'Apache-2.0', 'n_tokens_mistral': 719, 'n_tokens_neox': 709, 'n_words': 284, 'path': 'python/ejercicios/houses/batch_client.ipynb', 'repo_name': 'fhuertas/uah-mbi-2019-streaming', 'size': '2247'}