🔒 Mutliprocessing destroy saving data.

[lunar sable]

Hello, I would need some advice on how I could use multiprocessing in my use case.

Basically, my program grabs a list of keywords from an excel file. Then I want to start 50 tasks in parallel( multiprocessing) to process each of these keywords).
Then each process will save up the result in a csv file independantly. So like once the process finished, it willl save it.

This caused issues when I was using multiprocessing, because sometimes they would try to write the file simultaneously and it breaks the data.
Notice in the picture how after 800 successful data insertion, it just broke for no reason.
Thank you for any help

[peak hamlet]

@lunar sable

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.

[formal viper]

show the code how you save to the CSV file

it looks almost correct but looks like the fields are not quoted, so excel (or whatever program you're using) will display it as separate columns

[lunar sable]

    async def saveResult(self, finalText, title, keyword):
        parent_directory = os.path.basename(
            os.path.dirname(self.KWfile))

        # Specify the folder where you want to save the file
        output_folder = f'results/{str(parent_directory)}.csv'

        # Check if the file exists
        if not os.path.exists(output_folder):
            # If it doesn't exist, create and write to it
            data = [
                ['Title', 'Description', 'Status', 'Categories', 'tags'],
                [title, finalText, 'draft', 'Golden Retriever', keyword]

            ]

            with open(output_folder, 'w', newline='', encoding='utf-8') as csv_file:
                csv_writer = csv.writer(csv_file)
                csv_writer.writerows(data)

            print(f'CSV file "{output_folder}" created successfully.')
        else:
            with open(output_folder, 'a', newline='', encoding='utf-8') as csv_file:
                data = [
                    [title, finalText, 'draft', 'Golden Retriever', keyword]
                ]
                csv_writer = csv.writer(csv_file)
                csv_writer.writerows(data)

            print(f'CSV file "{output_folder}" already exists.')

[lunar sable]

it displayed the first 800 entries correctly, then it broke.

[formal viper]

are you writing into one file?

or one CSV per process?

[lunar sable]

into one file

All processed write to the same file

[formal viper]

there we go

don't do that

[lunar sable]

but I have to

I need to deliver a single csv file. Not 50

[formal viper]

write one CSV per process and then join them OR keep the data in memory and only write to CSV once, when you have all the data

[lunar sable]

problem is that processing takes like a lot of time and money. If there is an error after 2 hours and it doesnt save, I would lose money for nothing

so writing immediately is safe for me

[oak gale]

You could use a queue of what you want to write and use a coordinating process to do the writes

[lunar sable]

like multiprocessing.queue?

But can processes share data in a single queue?

[formal viper]

yes, it's possible

[oak gale]

Like use multiprocessing.Pool.imap_unordered

And return what you want to write to your csv from your mapper function

[lunar sable]

Would need to look into it, but the solution of writing one CSV per process and joining after seems the easiest to me. I kind of have no idea how to implement a queue. Honestly, I am bit new with the whole multiprocessing system.

[oak gale]

imap_unordered does the queuing for you

[lunar sable]

this is that runs the multiprocessing:

    def run(self):

        first_column_data = asyncio.run(self.getQueue())
        print(first_column_data)
        self.getProxies()
        num_processes = 50
        pool = multiprocessing.Pool(processes=num_processes)
        for i, keyword in enumerate(first_column_data):
            if "None" not in str(keyword):
                time.sleep(1.25)  # Introduce a delay of 1 second
                pool.apply_async(self.process_keyword, args=(keyword,))

        pool.close()
        pool.join()

I just change multiprocessing.Pool with multiprocessing.Pool.imap_unordered?

[oak gale]

Instead of pool.apply_async you pass a generator comprehension to pool.imap_unordered

https://docs.python.org/3/library/multiprocessing.html#multiprocessing.pool.Pool.imap_unordered

for result in pool.imap_unordered(self.process_keyword, first_column_data):
    csv_writer.writerows(result)

And return the rows you went to write from process_keyword

[lunar sable]

This is initially how it looks like

    def process_keyword(self, keyword):
        proxy = random.choices(self.proxies)
        if keyword[1] is not None:
            print(keyword[1])
            thread = MistralAG(proxy[0], keyword[1])
            final_text, title = thread.run()
            if final_text == "Failure":
                asyncio.run(self.saveProgress(keyword[0]))
            else:
                asyncio.run(self.saveResult(final_text, title, keyword[1]))
                asyncio.run(self.saveProgress(keyword[0]))
        print("Stop")

    def run(self):

        first_column_data = asyncio.run(self.getQueue())
        print(first_column_data)
        self.getProxies()
        num_processes = 1
        pool = multiprocessing.Pool(processes=num_processes)
        for i, keyword in enumerate(first_column_data):
            if "None" not in str(keyword):
                time.sleep(1.25)  # Introduce a delay of 1 second
                print(keyword)
                pool.imap_unordered(self.process_keyword, keyword)

        pool.close()
        pool.join()

Should I make it look like this:

       def run(self):

        first_column_data = asyncio.run(self.getQueue())
        print(first_column_data)
        self.getProxies()
        num_processes = 1
        pool = multiprocessing.Pool(processes=num_processes)
        for i, keyword in enumerate(first_column_data):
            if "None" not in str(keyword):
                time.sleep(1.25)  # Introduce a delay of 1 second

                pool.imap_unordered(self.process_keyword, [keyword])

Like so?
The saving is done in process_keyword function

After it does saveResult

[oak gale]

You need to iterate the imap_unordered and save the results from process_keyword

And process_keyword needs to return the results instead of saving them

[lunar sable]

Alright, will do these modifications.

[formal viper]

also you can use the mp pool as a context processor and remove the join and close at the end:

with multiprocessing.Pool(...) as pool:
    # code

[oak gale]

And you need to pass first_column_data to imap_unordered not [keyword]

[lunar sable]

Took me some time to figure it out, but should it look like this?

    def run(self):

        first_column_data = asyncio.run(self.getQueue())
        print(first_column_data)
        self.getProxies()
        num_processes = 1
        real_data = []
        pool = multiprocessing.Pool(processes=num_processes)
        for i, keyword in enumerate(first_column_data):
            if "None" not in str(keyword):
                real_data.append(keyword) # Introduce a delay of 1 second
        print(real_data)

        for (FinalText, thetitle, keyword) in pool.imap_unordered(self.process_keyword, [real_data]):
            if FinalText == "Failure":
                asyncio.run(self.saveProgress(keyword))
            else:
                asyncio.run(self.saveResult(FinalText, thetitle, keyword))
                asyncio.run(self.saveProgress(keyword))

        pool.close()
        pool.join()

[lunar sable]

oh yeah, will do modify that too

[oak gale]

You need to pass real_data not [real_data]

[lunar sable]

Okay, I think all modifications are done and it succeeded into doing one run.

So now it means that the csv saving issue should be fixed permanently?

Thank you a lot for your patience and help, it is greatly appreciated.

[oak gale]

It might be neater to use a comprehension to produce your real_data

[lunar sable]

will do, I just hope I can fix this issue and finally start a full run.

thank you a lot

[lunar sable]

also, is it possible to add a small delay between processes. Like a 1.25 second delay, so it doesnt run in the same time

[oak gale]

Not easily

What's the problem when there's no delay?

[lunar sable]

I do a lot of api calls, and I want to avoid to be rate limited when I use multiple processes.

when adding a little delay, it gives some spaces

For example, I have 200 maximum concurrent requests. When adding a delay, it gives time for the previous request to finish

like a little trick to increase the concurrency and speed

[oak gale]

Do you even need multiprocessing if you're doing everything with asyncio? Do you have blocking code?

Then you can use an asyncio.Semaphore

[lunar sable]

I think I am not very sure to know how to differentiate blocking code. Like, I do not have any file reading or time.sleep when doing the run. Only at the end when it saves the result

[oak gale]

Probably 200 concurrent requests will be fine without multiprocessing on one thread

[lunar sable]

So it will cost less ressources?

I see. Then its worth looking into it

I might need a little help on how to implement it. Do I just need to replace the multiprocess.pool with asyncio.Semaphore

Do I have to do some other modifications in my other scripts related to the run?

[oak gale]

No I'd probably use aiometer

[lunar sable]

Its getting overwhelming for me. I thought of this solution.
time.sleep( 1.25 * ProcessNumber)

I think it should work, its a nice hack

[peak hamlet]

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.