[Python] How to store data collected with kafka in Hdfs?

I have a school project that consist to collect data using kafka, store them in hdfs for later analyse with spark. I already collect the data with kafka and now I am trying to store them in hdfs, but I have an issue:

Erreur lors de l'enregistrement dans HDFS : HTTPConnectionPool(host='hadoop-worker2', port=9864): Max retries exceeded with url: /webhdfs/v1/data/openAQ/openAQ_data.json?op=APPEND&user.name=hdfs&namenoderpcaddress=hadoop-master:9000&user.name=hdfs (Caused by NameResolutionError("<urllib3.connection.HTTPConnection object at 0x00000276542F1700>: Failed to resolve 'hadoop-worker2' ([Errno 11001] getaddrinfo failed)"))

Here is the code I implement:

from hdfs import InsecureClient


#### Déclaration des variables

KAFKA_BROKER = "localhost:9092"
KAFKA_TOPIC = "BigDataProjet"

# Configuration du client HDFS
HDFS_URL = "http://localhost:9870" # URL du Namenode
HDFS_PATH = "/data/openAQ"
HDFS_FILE_PATH = f"{HDFS_PATH}/openAQ_data.json"

client = InsecureClient(HDFS_URL, user="hdfs")


# Initialisation du consumer kafka

consumer = KafkaConsumer(
KAFKA_TOPIC,
bootstrap_servers=KAFKA_BROKER,
auto_offset_reset="earliest",
enable_auto_commit=True
)


def sauvegarder_hdfs(data):
""" Sauvegarde les données dans un fichier HDFS """
try:
# Vérifier si le répertoire existe, sinon le créer
if not client.status(HDFS_PATH, strict=False):
client.makedirs(HDFS_PATH)

# Écrire les données en mode append
with client.write(HDFS_FILE_PATH, encoding='utf-8', append=True) as writer:
writer.write(data + "\n")
print("Données enregistrées dans HDFS")
except Exception as e:
print(f"Erreur lors de l'enregistrement dans HDFS : {e}")

def consumer_kafka():
""" Lit les messages Kafka et les stocke dans HDFS """
print("En attente des données Kafka...")
for message in consumer:
data = message.value.decode("utf-8")
print(f"données correctement récuperer: {data}")
sauvegarder_hdfs(data)

if __name__ == "__main__":
consumer_kafka()

Continue reading...