Can't achieve desired directory structure when writing data from Kafka topic to HDFS using PySpark

I am reading data from a Kafka topic and storing the data to HDFS using PySpark. I am trying to achieve the following directory structure in HDFS when I run the code:

user/hadoop/OUTPUT_FINAL1/year/month/day/hour/
csv/ (all csv files as parquet)
images/ (all images files)
videos/ (all videos files)

However when I run my code I the directory structure is seen as the following in HDFS:

user/hadoop/OUTPUT_FINAL1/year/month/day/hour/

csv/file_type=csv/ (csv data) and file_type=image/ (images data)

My code is not writing the desired directory structure and it is not writing the video files.

CSV files need to be written with the same name with the extension parquet.
Images need to be written with the the .jpg or .png extension and same for videos with .mp4 or .mov.

here is my code.

import logging
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
import pyarrow.hdfs as hdfs
import os
# Configure logging
logging.getLogger("py4j").setLevel(logging.ERROR)
logging.getLogger("org.apache.kafka").setLevel(logging.WARN)
# Initialize SparkSession
spark = SparkSession.builder \
.appName("KafkaToHDFS") \
.config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.4.0") \
.getOrCreate()
# Kafka broker configuration
kafka_server = "sandbox-hdf.hortonworks.com:6667"
kafka_topic = "airbnbdata-final"
# Read data from Kafka topic
df = spark.read \
.format("kafka") \
.option("kafka.bootstrap.servers", kafka_server) \
.option("subscribe", kafka_topic) \
.load()
# Define a function to create the output directory
def create_directory(directory):
current_time = spark.sql("SELECT current_timestamp()").collect()[0][0]
# Extract year, month, day, and hour components
year_str = current_time.strftime("%Y")
month_str = current_time.strftime("%m")
day_str = current_time.strftime("%d")
hour_str = current_time.strftime("%H")
fulldirectory = directory + "/" + year_str + "/" + month_str + "/" + day_str + "/" + hour_str + "/"
return fulldirectory
# Define a function to process each batch
def process_batch(df, fulldirectory):
schema = StructType().add("file_type", StringType()).add("file_name", StringType()).add("data", StringType())
# Extract the key, value, and other columns
extracted_df = df.selectExpr(
"CAST(key AS STRING)", "CAST(value AS STRING)", "topic", "partition", "offset", "timestamp", "timestampType"
)
# Parse the JSON string in the value column
extracted_df = extracted_df.withColumn("value_json", from_json(col("value"), schema))
# Extract fields from the value_json column
extracted_df = extracted_df.withColumn("file_type", col("value_json.file_type"))
extracted_df = extracted_df.withColumn("file_name", col("value_json.file_name"))
extracted_df = extracted_df.withColumn("data", col("value_json.data"))
extracted_df.printSchema()
return extracted_df
# Define a custom function to write each file to HDFS
def write_file(row, fulldirectory):
fs = hdfs.connect(host="sandbox-hdp.hortonworks.com", port=8020)
file_name = row["file_name"]
data = row["data"]
hdfs_path = fulldirectory + file_name
with fs.open(hdfs_path, "wb") as f:
f.write(data)
# Define a function to process images and videos
def process_images_and_videos(df, fulldirectory):
# Filter the dataframe to get only images and videos
image_df = df.filter(col("file_type").isin(["jpg", "png"]))
video_df = df.filter(col("file_type").isin(["mp4", "mov"]))
# Write images to HDFS
image_df.foreach(lambda row: write_file(row, fulldirectory + "images/"))
# Write videos to HDFS
video_df.foreach(lambda row: write_file(row, fulldirectory + "videos/"))
# Configure the directory path and create the output directory
directory_path = "hdfs://sandbox-hdp.hortonworks.com:8020/user/hadoop/OUTPUT_FINAL1"
fulldirectory = create_directory(directory_path)
# Process the batch
processed_df = process_batch(df, fulldirectory)
# Write each batch to HDFS as Parquet
processed_df.write \
.format("parquet") \
.mode("append") \
.partitionBy("file_type", "file_name") \
.parquet(fulldirectory + "csv/")
# Process images and videos
process_images_and_videos(processed_df, fulldirectory)
# Stop the SparkSession
spark.stop()

I run my code in my ambari sandbox environment ssh using the following command:

spark-submit --jars /tmp/Airbnb_Data/jsr305-3.0.2.jar,/tmp/Airbnb_Data/snappy-java-1.1.8.4.jar,/tmp/Airbnb_Data/kafka-clients-3.3.2.jar,/tmp/Airbnb_Data/spark-sql-kafka-0-10_2.11-2.4.0.jar,/tmp/Airbnb_Data/commons-logging-1.1.3.jar,/tmp/Airbnb_Data/spark-streaming-kafka-0-10_2.12-3.4.0.jar,/tmp/Airbnb_Data/hadoop-client-api-3.3.4.jar,/tmp/Airbnb_Data/lz4-java-1.8.0.jar,/tmp/Airbnb_Data/hadoop-client-runtime-3.3.4.jar,/tmp/Airbnb_Data/scala-library-2.12.17.jar,/tmp/Airbnb_Data/spark-tags_2.12-3.4.0.jar,/tmp/Airbnb_Data/spark-token-provider-kafka-0-10_2.12-3.4.0.jar,/tmp/Airbnb_Data/slf4j-api-1.7.36.jar KafkaToHDFS.py

Here is my desired directory structure:

user/hadoop/OUTPUT_FINAL1/
├── year/
│   ├── month/
│   │   ├── day/
│   │   │   ├── hour/
│   │   │   │   ├── csv/
│   │   │   │   │   └── <csv parquet files>
│   │   │   │   ├── images/
│   │   │   │   │   └── <image files>
│   │   │   │   └── videos/
│   │   │   │       └── <video files>

Spark cannot write binary image or video data.

The fact that you're calling pyarrow functions within Spark foreach is not truly using Spark filesystem writer. Plus, pyarrow was released after HDP 2.6.5 support ended, I believe, so I don't think that function will work.

Regarding the directory structure, let Spark do that for you by partitioning the individual date columns rather than collecting the batch and extracting one element.

Then you'll have year=xxxx/month=xx/day=xx/hour=xx/filetype=xxx/ data

> Images need to be written with the the .jpg or .png extension and same for videos with .mp4 or .mov.

HDFS is not a good location for such files. Use blob storage for media assets