Interprocess communication (IPC) exchanges data or coordination signals between independent processes. These examples use a local UNIX socket to connect processes to Arrow’s Plasma shared-memory server, and queues to pass jobs and results. section 7.5.2 section 3.5.1
Two patterns illustrate different IPC needs:
- Plasma stores named objects in shared memory, letting a consumer find data even if the producer has exited. It is local rather than network-accessible; insufficient memory can cause older objects to be evicted. section 7.5.2
- Multiprocessing queues carry work and results between server components. They synchronize access, but only accept picklable objects; serialization can make communication slow. Bounded queues also require handling possible submission limits. section 3.5.1
plasma_store -s /tmp/fast_python -m 1000000000
Start the local Plasma server with a UNIX socket and a one-gigabyte shared-memory allocation. section 7.5.2
import os
import sys
import pyarrow as pa
from pyarrow import csv
import pyarrow.plasma as plasma
csv_name = sys.argv[1]
client = plasma.connect("/tmp/fast_python") #1
convert_options = csv.ConvertOptions( #2
column_types={
"VendorID": pa.bool_()
},
true_values=["Y", "1"],
false_values=["N", "2"])
table = csv.read_csv(
csv_name
convert_options=convert_options
)
pid = os.getpid()
plid = plasma.ObjectID(
f"csv-{pid}".ljust(20, " ").encode("us-ascii")) #3
client.put(table, plid) #4
The producer connects to Plasma, reads a CSV with Arrow, assigns the table an object ID, and stores it for other processes. section 7.5.2
import asyncio
from queue import Queue, Empty
import multiprocessing as mp
import types
work_queue = Queue()
results_queue = Queue()
results = {}
def worker(): #1
pool = mp.Pool() #2
while True:
job_id, code, data = work_queue.get() #3
func = types.FunctionType(code, globals(), 'mapper_and_reducer')
mapper, reducer = func()
counts = mr.map_reduce(pool, data, mapper, reducer, 100, mr.reporter)
results_queue.put((job_id, counts)) #4
pool.close()
pool.join()
async def main():
server = await asyncio.start_server(accept_requests, '127.0.0.1', 1936)
worker_thread = threading.Thread(target=worker) #5
worker_thread.start() #6
async with server:
await server.serve_forever()
asyncio.run(main())
A worker thread receives jobs from a queue, runs a multiprocessing pool, and places results on a second queue. section 3.5.1
