🔒 Facing Issue with Pyspark and repartition

[gaunt helm]

@hard saddle

Python help channel opened

Remember to:

• Ask your Python question, not if you can ask or if there's an expert who can help.
• Show a code sample as text (rather than a screenshot) and the error message, if you've got one.
• Explain what you expect to happen and what actually happens.

:warning: Do not pip install anything that isn't related to your question, especially if asked to over DMs.

Closes after a period of inactivity, or when you send !close.

[hard saddle]

this is my code

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from abc import ABC, abstractmethod
import threading


class ThreadedTask(ABC):
    @abstractmethod
    def process_chunk(self, data):
        pass


class Subclass1(ThreadedTask):
    def process_chunk(self, data):
        print("Subclass 1 processing chunk")
        data.show()

class Subclass2(ThreadedTask):
    def process_chunk(self, data):
        print("Subclass 2 processing chunk")
        data.groupBy(col("column_name")).count().show()


class Subclass3(ThreadedTask):
    def process_chunk(self, data):
        print("Subclass 3 processing chunk")

class CsvSplitterProcessor:
    def __init__(self, spark, csv_path, num_chunks):
        self.spark = spark
        self.csv_path = csv_path
        self.num_chunks = num_chunks
        self.subclasses = [Subclass1(), Subclass2(), Subclass3()]

    def process(self):
        print("Starting processing")
        data = self.spark.read.csv(self.csv_path)
        data_rdd = data.rdd
        split_data = data_rdd.repartition(self.num_chunks)
        def process_chunk_iterator(iterator):
            dataframe = self.spark.createDataFrame(iterator)
            for subclass in self.subclasses:
                subclass.process_chunk(dataframe)
        split_data.foreachPartition(process_chunk_iterator)
        print(split_data.getNumPartitions())

    def process_chunk(self, iterator):
        dataframe = self.spark.createDataFrame(
            iterator)  # Create DataFrame within partition
        print(dataframe.show())
        for subclass in self.subclasses:
            subclass.process_chunk(dataframe)

spark = SparkSession.builder.appName("2").getOrCreate()
csv_path = r"\input\data.csv"
processor = CsvSplitterProcessor(
    spark, csv_path, 3)
processor.process()
spark.stop()

def process(self):
data = self.spark.read.csv(self.csv_path)
# Convert DataFrame to RDD for repartitioning
data_rdd = data.rdd
split_data = data_rdd.repartition(self.num_chunks)

# Process each partition using foreachPartition
split_data.foreachPartition(self.process_chunk)

# Add an action to trigger execution (optional)
split_data.count()  # Example action

##############

def process_chunk(self, iterator):
print("Processing chunk started...") # Logging for verification
dataframe = self.spark.createDataFrame(iterator)
# Access SparkSession within the subclass if needed
for subclass in self.subclasses:
subclass.process_chunk(dataframe)
I get the error on this line
split_data.foreachPartition(self.process_chunk)
Cuz the print on the procces_chunk method not work

[gaunt helm]

@hard saddle

Python help channel closed

This help channel has been closed and it's no longer possible to send messages here. If your question wasn't answered, feel free to create a new post in #1035199133436354600. To maximize your chances of getting a response, check out this guide on asking good questions.