Arquivo de Paralela - Área de Trampo

Dask Paralelização de Dados

22/05/202622/04/2026 Por antonino

Dask é uma biblioteca para computação paralela em Python. Ele escala de um laptop para um cluster de milhares de máquinas. Primeiramente, Dask integra-se perfeitamente com NumPy e Pandas. Por exemplo, você processa DataFrames maiores que a memória RAM. Além disso, Dask usa execução lazy (preguiçosa) para otimização. Assim, o sistema constrói um grafo antes de qualquer cálculo. Consequentemente, operações desnecessárias são evitadas. Quando utilizar Dask? Em dados que não cabem na memória. Também em computação paralela com APIs familiares. Por outro lado, para dados pequenos, Dask adiciona overhead. Dask oferece arrays, dataframes, bags e delayed. Então, vamos explorar cada um com exemplos práticos. Três subtítulos guiarão você pelo universo Dask. Portanto, ao final, você processará terabytes de dados.

Dask arrays: numpy que escala

Dask arrays imitam NumPy, mas trabalham com blocos (chunks). Eles dividem grandes arrays em pedaços menores. Cada bloco processa separadamente e em paralelo. Quando usar Dask arrays? Em operações matemáticas em dados grandes. Por exemplo, multiplicação de matrizes com bilhões de elementos. Além disso, você controla o tamanho dos chunks manualmente. Exemplo de Dask array:

import dask.array as da
import numpy as np
import time

# Criando um array gigante (10.000 x 10.000 = 100 milhões de elementos)
# Isso não caberia na memória se fosse NumPy puro
tamanho = 10000
print(f"Criando array de {tamanho}x{tamanho} ({tamanho*tamanho/1e6:.0f}M elementos)")

# Com NumPy (carrega tudo na memória)
inicio_np = time.time()
arr_np = np.random.random((tamanho, tamanho))
soma_np = arr_np.sum()
tempo_np = time.time() - inicio_np
print(f"NumPy: soma={soma_np:.2f}, tempo={tempo_np:.2f}s, memória={arr_np.nbytes/1e6:.0f}MB")

# Com Dask (processa em blocos)
inicio_dask = time.time()
# Blocos de 1000x1000 (100 blocos no total)
arr_dask = da.random.random((tamanho, tamanho), chunks=(1000, 1000))
soma_dask = arr_dask.sum().compute()  # compute() executa o grafo
tempo_dask = time.time() - inicio_dask
print(f"Dask: soma={soma_dask:.2f}, tempo={tempo_dask:.2f}s")

# Operações matemáticas com Dask
arr1 = da.ones((10000, 10000), chunks=(1000, 1000))
arr2 = da.ones((10000, 10000), chunks=(1000, 1000))

# Multiplicação elemento a elemento (lazy)
resultado = (arr1 + arr2) * 3 - 1
print(f"\nOperação lazy criada, shape: {resultado.shape}")

# Calcula a média (executa o grafo)
media = resultado.mean().compute()
print(f"Média do resultado: {media:.2f}")

# Demonstração do grafo
print(f"Gráfico de tarefas: {resultado}")

import dask.array as da

import numpy as np

import time

# Criando um array gigante (10.000 x 10.000 = 100 milhões de elementos)

# Isso não caberia na memória se fosse NumPy puro

tamanho = 10000

print(f"Criando array de {tamanho}x{tamanho} ({tamanho*tamanho/1e6:.0f}M elementos)")

# Com NumPy (carrega tudo na memória)

inicio_np = time.time()

arr_np = np.random.random((tamanho, tamanho))

soma_np = arr_np.sum()

tempo_np = time.time() - inicio_np

print(f"NumPy: soma={soma_np:.2f}, tempo={tempo_np:.2f}s, memória={arr_np.nbytes/1e6:.0f}MB")

# Com Dask (processa em blocos)

inicio_dask = time.time()

# Blocos de 1000x1000 (100 blocos no total)

arr_dask = da.random.random((tamanho, tamanho), chunks=(1000, 1000))

soma_dask = arr_dask.sum().compute() # compute() executa o grafo

tempo_dask = time.time() - inicio_dask

print(f"Dask: soma={soma_dask:.2f}, tempo={tempo_dask:.2f}s")

# Operações matemáticas com Dask

arr1 = da.ones((10000, 10000), chunks=(1000, 1000))

arr2 = da.ones((10000, 10000), chunks=(1000, 1000))

# Multiplicação elemento a elemento (lazy)

resultado = (arr1 + arr2) * 3 - 1

print(f"\nOperação lazy criada, shape: {resultado.shape}")

# Calcula a média (executa o grafo)

media = resultado.mean().compute()

print(f"Média do resultado: {media:.2f}")

# Demonstração do grafo

print(f"Gráfico de tarefas: {resultado}")

Dask processa arrays maiores que a memória RAM. Ele usa disco ou cluster conforme a necessidade. Portanto, você nunca enfrenta MemoryError novamente.

Dask dataframes: pandas que escala

Dask DataFrames imitam pandas, mas particionados em pedaços. Cada partição contém um DataFrame pandas comum. Operações como groupby, join e merge são distribuídas. Quando usar Dask DataFrames? Em dados tabulares gigantes. Por exemplo, logs de servidor com bilhões de linhas. Além disso, a leitura de múltiplos arquivos ocorre em paralelo. Exemplo de Dask DataFrame:

import dask.dataframe as dd
import pandas as pd
import numpy as np
import time

# Criando um dataset grande simulado
print("=== Dask DataFrame vs Pandas ===\n")
n_linhas = 10_000_000  # 10 milhões de linhas
print(f"Simulando {n_linhas:,} linhas de dados")

# Versão Pandas (carrega tudo na memória)
try:
    inicio_pd = time.time()
    df_pd = pd.DataFrame({
        'id': range(n_linhas),
        'valor': np.random.random(n_linhas),
        'categoria': np.random.choice(['A', 'B', 'C', 'D'], n_linhas)
    })
    media_pd = df_pd.groupby('categoria')['valor'].mean()
    tempo_pd = time.time() - inicio_pd
    print(f"Pandas: média por categoria em {tempo_pd:.2f}s")
    print(f"  Resultado: {media_pd.to_dict()}")
except MemoryError:
    print("Pandas: Memória insuficiente para 10M linhas!")

# Versão Dask (processa em partições)
inicio_dd = time.time()
# Cria DataFrame Dask a partir de um gerador
df_dd = dd.from_pandas(
    pd.DataFrame({
        'id': range(100_000),
        'valor': np.random.random(100_000),
        'categoria': np.random.choice(['A', 'B', 'C', 'D'], 100_000)
    }), 
    npartitions=4
)

# Simula 100 repetições (total 10M linhas)
for _ in range(99):
    df_temp = dd.from_pandas(
        pd.DataFrame({
            'id': range(100_000),
            'valor': np.random.random(100_000),
            'categoria': np.random.choice(['A', 'B', 'C', 'D'], 100_000)
        }),
        npartitions=4
    )
    df_dd = dd.concat([df_dd, df_temp])

# Operação groupby lazy
media_dd = df_dd.groupby('categoria')['valor'].mean()
# Executa o cálculo
resultado_dd = media_dd.compute()
tempo_dd = time.time() - inicio_dd
print(f"Dask: média por categoria em {tempo_dd:.2f}s")
print(f"  Resultado: {resultado_dd.to_dict()}")

# Exemplo com leitura de CSV grande (simulado)
print("\n=== Leitura de CSV Particionado ===")
# Criando arquivos CSV simulados
import tempfile
import os

with tempfile.TemporaryDirectory() as tmpdir:
    for i in range(5):
        df_part = pd.DataFrame({
            'id': range(100_000),
            'valor': np.random.random(100_000),
            'categoria': np.random.choice(['X', 'Y', 'Z'], 100_000)
        })
        df_part.to_csv(f"{tmpdir}/part_{i}.csv", index=False)
    
    # Lê todos os CSVs de uma vez com Dask
    df_dask = dd.read_csv(f"{tmpdir}/part_*.csv")
    print(f"Partições: {df_dask.npartitions}")
    print(f"Colunas: {df_dask.columns.tolist()}")
    
    # Calcula estatísticas
    stats = df_dask.groupby('categoria')['valor'].agg(['mean', 'std', 'count'])
    resultado = stats.compute()
    print("\nEstatísticas por categoria:")
    print(resultado)

import dask.dataframe as dd

import pandas as pd

import numpy as np

import time

# Criando um dataset grande simulado

print("=== Dask DataFrame vs Pandas ===\n")

n_linhas = 10_000_000 # 10 milhões de linhas

print(f"Simulando {n_linhas:,} linhas de dados")

# Versão Pandas (carrega tudo na memória)

try:

inicio_pd = time.time()

df_pd = pd.DataFrame({

'id': range(n_linhas),

'valor': np.random.random(n_linhas),

'categoria': np.random.choice(['A', 'B', 'C', 'D'], n_linhas)

})

media_pd = df_pd.groupby('categoria')['valor'].mean()

tempo_pd = time.time() - inicio_pd

print(f"Pandas: média por categoria em {tempo_pd:.2f}s")

print(f" Resultado: {media_pd.to_dict()}")

except MemoryError:

print("Pandas: Memória insuficiente para 10M linhas!")

# Versão Dask (processa em partições)

inicio_dd = time.time()

# Cria DataFrame Dask a partir de um gerador

df_dd = dd.from_pandas(

pd.DataFrame({

'id': range(100_000),

'valor': np.random.random(100_000),

'categoria': np.random.choice(['A', 'B', 'C', 'D'], 100_000)

}),

npartitions=4

)

# Simula 100 repetições (total 10M linhas)

for _ in range(99):

df_temp = dd.from_pandas(

pd.DataFrame({

'id': range(100_000),

'valor': np.random.random(100_000),

'categoria': np.random.choice(['A', 'B', 'C', 'D'], 100_000)

}),

npartitions=4

)

df_dd = dd.concat([df_dd, df_temp])

# Operação groupby lazy

media_dd = df_dd.groupby('categoria')['valor'].mean()

# Executa o cálculo

resultado_dd = media_dd.compute()

tempo_dd = time.time() - inicio_dd

print(f"Dask: média por categoria em {tempo_dd:.2f}s")

print(f" Resultado: {resultado_dd.to_dict()}")

# Exemplo com leitura de CSV grande (simulado)

print("\n=== Leitura de CSV Particionado ===")

# Criando arquivos CSV simulados

import tempfile

import os

with tempfile.TemporaryDirectory() as tmpdir:

for i in range(5):

df_part = pd.DataFrame({

'id': range(100_000),

'valor': np.random.random(100_000),

'categoria': np.random.choice(['X', 'Y', 'Z'], 100_000)

})

df_part.to_csv(f"{tmpdir}/part_{i}.csv", index=False)

# Lê todos os CSVs de uma vez com Dask

df_dask = dd.read_csv(f"{tmpdir}/part_*.csv")

print(f"Partições: {df_dask.npartitions}")

print(f"Colunas: {df_dask.columns.tolist()}")

# Calcula estatísticas

stats = df_dask.groupby('categoria')['valor'].agg(['mean', 'std', 'count'])

resultado = stats.compute()

print("\nEstatísticas por categoria:")

print(resultado)

Dask DataFrames processam dados maiores que a memória. Eles também leem múltiplos arquivos em paralelo. Assim, você trabalha com terabytes como se fossem gigabytes.

Dask delayed: paralelização customizada

Dask delayed decora funções para execução lazy e paralela. Você controla exatamente o grafo de dependências. Quando usar delayed? Em fluxos de trabalho complexos. Por exemplo, pipelines de ETL com etapas interdependentes. Além disso, delayed permite paralelismo granular e fino. Exemplo de Dask delayed:

import dask
from dask.distributed import Client
import time
import random

# Inicia cliente Dask local
client = Client(n_workers=4, threads_per_worker=1)
print(f"Dashboard Dask: {client.dashboard_link}")

# Funções de exemplo (simulam trabalho)
@dask.delayed
def baixar_dados(url):
    """Simula download de dados."""
    time.sleep(random.uniform(0.5, 1.5))
    return f"Dados de {url}"

@dask.delayed
def processar_dados(dados):
    """Processa os dados baixados."""
    time.sleep(0.5)
    return f"Processado: {dados}"

@dask.delayed
def analisar_dados(dados_processados):
    """Analisa os dados processados."""
    time.sleep(0.3)
    return f"Análise: {len(dados_processados)} caracteres"

@dask.delayed
def salvar_resultado(analise):
    """Salva o resultado final."""
    time.sleep(0.2)
    return f"Salvo: {analise}"

# Pipeline completo com Dask delayed
print("=== Pipeline de Dados com Dask Delayed ===\n")

urls = [f"http://api.com/dados/{i}" for i in range(10)]

# Cria o grafo lazy (nenhuma execução ainda)
inicio_pipeline = time.time()

# Passo 1: Baixar todos os dados em paralelo
downloads = [baixar_dados(url) for url in urls]

# Passo 2: Processar cada download
processados = [processar_dados(d) for d in downloads]

# Passo 3: Analisar cada resultado
analises = [analisar_dados(p) for p in processados]

# Passo 4: Salvar todos os resultados
salvos = [salvar_resultado(a) for a in analises]

# Executa o grafo (agora sim)
resultados = dask.compute(*salvos)
tempo_pipeline = time.time() - inicio_pipeline

print(f"Pipeline concluído em {tempo_pipeline:.2f}s")
for i, r in enumerate(resultados[:5]):
    print(f"  Resultado {i}: {r}")

# Exemplo com dependência entre tarefas
print("\n=== Dependências entre Tarefas ===")

@dask.delayed
def etapa1():
    time.sleep(0.5)
    return 10

@dask.delayed
def etapa2(x):
    time.sleep(0.5)
    return x * 2

@dask.delayed
def etapa3(x, y):
    time.sleep(0.5)
    return x + y

# Grafo com dependências (etapa2 depende de etapa1)
a = etapa1()
b = etapa2(a)  # depende de a
c = etapa2(5)  # independente
d = etapa3(b, c)  # depende de b e c

# Executa
resultado_final = dask.compute(d)
print(f"Resultado final: {resultado_final[0]}")

# Exemplo de visualização do grafo (opcional)
print("\n=== Visualização (opcional) ===")
print("Para ver o grafo: dask.visualize(d, filename='grafo.png')")

# Fecha o cliente
client.close()

import dask

from dask.distributed import Client

import time

import random

# Inicia cliente Dask local

client = Client(n_workers=4, threads_per_worker=1)

print(f"Dashboard Dask: {client.dashboard_link}")

# Funções de exemplo (simulam trabalho)

@dask.delayed

def baixar_dados(url):

"""Simula download de dados."""

time.sleep(random.uniform(0.5, 1.5))

return f"Dados de {url}"

@dask.delayed

def processar_dados(dados):

"""Processa os dados baixados."""

time.sleep(0.5)

return f"Processado: {dados}"

@dask.delayed

def analisar_dados(dados_processados):

"""Analisa os dados processados."""

time.sleep(0.3)

return f"Análise: {len(dados_processados)} caracteres"

@dask.delayed

def salvar_resultado(analise):

"""Salva o resultado final."""

time.sleep(0.2)

return f"Salvo: {analise}"

# Pipeline completo com Dask delayed

print("=== Pipeline de Dados com Dask Delayed ===\n")

urls = [f"http://api.com/dados/{i}" for i in range(10)]

# Cria o grafo lazy (nenhuma execução ainda)

inicio_pipeline = time.time()

# Passo 1: Baixar todos os dados em paralelo

downloads = [baixar_dados(url) for url in urls]

# Passo 2: Processar cada download

processados = [processar_dados(d) for d in downloads]

# Passo 3: Analisar cada resultado

analises = [analisar_dados(p) for p in processados]

# Passo 4: Salvar todos os resultados

salvos = [salvar_resultado(a) for a in analises]

# Executa o grafo (agora sim)

resultados = dask.compute(*salvos)

tempo_pipeline = time.time() - inicio_pipeline

print(f"Pipeline concluído em {tempo_pipeline:.2f}s")

for i, r in enumerate(resultados[:5]):

print(f" Resultado {i}: {r}")

# Exemplo com dependência entre tarefas

print("\n=== Dependências entre Tarefas ===")

@dask.delayed

def etapa1():

time.sleep(0.5)

return 10

@dask.delayed

def etapa2(x):

time.sleep(0.5)

return x * 2

@dask.delayed

def etapa3(x, y):

time.sleep(0.5)

return x + y

# Grafo com dependências (etapa2 depende de etapa1)

a = etapa1()

b = etapa2(a) # depende de a

c = etapa2(5) # independente

d = etapa3(b, c) # depende de b e c

# Executa

resultado_final = dask.compute(d)

print(f"Resultado final: {resultado_final[0]}")

# Exemplo de visualização do grafo (opcional)

print("\n=== Visualização (opcional) ===")

print("Para ver o grafo: dask.visualize(d, filename='grafo.png')")

# Fecha o cliente

client.close()

Dask delayed é perfeito para pipelines personalizados. A fórmula da aceleração teórica funciona bem: \(S = \frac{T_{\text{seq}}}{T_{\text{dask}}} \approx N_{\text{workers}} \times (1 – O)\) Dask é a ferramenta ideal para Big Data em Python. Comece com Dask arrays para dados numéricos. Use Dask DataFrames para dados tabulares. Para fluxos complexos, Dask delayed é a escolha certa. Portanto, escalabilidade sem complicação é o poder do Dask.

Processos Paralelos em Python: Paralelismo Verdadeiro

22/05/202622/04/2026 Por antonino

Processos paralelos executam código simultaneamente em múltiplos núcleos. Isso é chamado de true parallelism ou paralelismo verdadeiro. Primeiramente, cada processo tem seu próprio interpretador e memória. Por exemplo, 8 processos podem rodar em 8 núcleos de CPU ao mesmo tempo. Além disso, processos não compartilham o GIL (Global Interpreter Lock). A voz passiva é usada aqui: “o sistema operacional gerencia a distribuição entre núcleos”. Quando utilizar processos paralelos? Em tarefas com CPU intensiva. Por exemplo, processamento de imagens, simulações científicas ou machine learning. Também em qualquer tarefa que exija todo o poder da máquina. Python oferece o módulo multiprocessing e o concurrent.futures. Vamos explorar características, ganhos e limitações. Três subtítulos guiarão você pelo verdadeiro paralelismo. Ao final, você dominará a execução paralela em Python.

Multiprocessing vs. threading: a diferença crucial

Threads são leves, mas limitadas pelo GIL em CPU-bound. Processos são mais pesados, mas executam em paralelo verdadeiro. Quando escolher processos? Quando o gargalo é CPU, não I/O. A criação de um processo é mais cara que uma thread. No entanto, o ganho em paralelismo compensa para tarefas longas. A voz passiva é aplicada: “a memória é duplicada para cada processo”. Exemplo comparativo entre threads e processos para CPU-bound:

import multiprocessing
import threading
import time
import os

def trabalho_cpu_pesado(segundos):
    """Função que consome muita CPU."""
    pid = os.getpid()
    inicio = time.time()
    # Loop pesado para simular trabalho
    total = 0
    for i in range(30_000_000):
        total += i * i
    fim = time.time()
    print(f"Processo/Thread PID {pid} concluiu em {fim - inicio:.2f}s")
    return total

def executar_com_processos(n_tarefas):
    """Executa com processos paralelos."""
    with multiprocessing.Pool(processes=n_tarefas) as pool:
        resultados = pool.map(trabalho_cpu_pesado, [1] * n_tarefas)
    return resultados

def executar_com_threads(n_tarefas):
    """Executa com threads (limitado pelo GIL)."""
    threads = []
    resultados = [0] * n_tarefas
    
    def wrapper(idx):
        resultados[idx] = trabalho_cpu_pesado(1)
    
    for i in range(n_tarefas):
        t = threading.Thread(target=wrapper, args=(i,))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    return resultados

if __name__ == "__main__":
    n = 4  # Número de tarefas (igual aos núcleos)
    
    print(f"Executando {n} tarefas CPU-bound com PROCESSOS:")
    inicio = time.time()
    executar_com_processos(n)
    tempo_processos = time.time() - inicio
    print(f"Tempo total com processos: {tempo_processos:.2f}s")
    
    print(f"\nExecutando {n} tarefas CPU-bound com THREADS:")
    inicio = time.time()
    executar_com_threads(n)
    tempo_threads = time.time() - inicio
    print(f"Tempo total com threads: {tempo_threads:.2f}s")
    
    print(f"\nProcessos foram {tempo_threads / tempo_processos:.1f}x mais rápidos!")
    print("(Processos executam em paralelo verdadeiro, threads não)")

import multiprocessing

import threading

import time

import os

def trabalho_cpu_pesado(segundos):

"""Função que consome muita CPU."""

pid = os.getpid()

inicio = time.time()

# Loop pesado para simular trabalho

total = 0

for i in range(30_000_000):

total += i * i

fim = time.time()

print(f"Processo/Thread PID {pid} concluiu em {fim - inicio:.2f}s")

return total

def executar_com_processos(n_tarefas):

"""Executa com processos paralelos."""

with multiprocessing.Pool(processes=n_tarefas) as pool:

resultados = pool.map(trabalho_cpu_pesado, [1] * n_tarefas)

return resultados

def executar_com_threads(n_tarefas):

"""Executa com threads (limitado pelo GIL)."""

threads = []

resultados = [0] * n_tarefas

def wrapper(idx):

resultados[idx] = trabalho_cpu_pesado(1)

for i in range(n_tarefas):

t = threading.Thread(target=wrapper, args=(i,))

threads.append(t)

t.start()

for t in threads:

t.join()

return resultados

if __name__ == "__main__":

n = 4 # Número de tarefas (igual aos núcleos)

print(f"Executando {n} tarefas CPU-bound com PROCESSOS:")

inicio = time.time()

executar_com_processos(n)

tempo_processos = time.time() - inicio

print(f"Tempo total com processos: {tempo_processos:.2f}s")

print(f"\nExecutando {n} tarefas CPU-bound com THREADS:")

inicio = time.time()

executar_com_threads(n)

tempo_threads = time.time() - inicio

print(f"Tempo total com threads: {tempo_threads:.2f}s")

print(f"\nProcessos foram {tempo_threads / tempo_processos:.1f}x mais rápidos!")

print("(Processos executam em paralelo verdadeiro, threads não)")

Processos mostram aceleração próxima ao número de núcleos. Threads mostram aceleração próxima de 1.0 devido ao GIL. Essa é a diferença fundamental entre concorrência e paralelismo.

Comunicação e sincronização entre processos

Processos paralelos precisam se comunicar para trocar resultados. Diferente de threads, eles não compartilham memória automaticamente. Portanto, usamos filas (Queue), pipes ou memória compartilhada. A serialização (pickle) é necessária para enviar dados entre processos. Isso adiciona overhead, mas é inevitável para isolamento. Quando usar memória compartilhada? Para grandes arrays numéricos. A voz passiva é usada aqui: “os dados são copiados por valor entre processos”. Exemplo de comunicação com Queue entre processos paralelos:

import multiprocessing
import time
import random

def produtor(queue, id_produtor, n_itens):
    """Produz itens e coloca na fila."""
    for i in range(n_itens):
        item = f"Produtor-{id_produtor}-Item-{i}"
        queue.put(item)
        print(f"[Produtor {id_produtor}] Produziu: {item}")
        time.sleep(random.uniform(0.1, 0.3))
    queue.put(None)  # Sinal de fim para este produtor

def consumidor(queue, id_consumidor):
    """Consome itens da fila até receber Nones suficientes."""
    n_finalizados = 0
    while True:
        item = queue.get()
        if item is None:
            n_finalizados += 1
            if n_finalizados == 2:  # 2 produtores
                break
            continue
        print(f"[Consumidor {id_consumidor}] Consumiu: {item}")
        # Simula processamento do item
        time.sleep(random.uniform(0.2, 0.5))

if __name__ == "__main__":
    queue = multiprocessing.Queue()
    
    # Criando produtores paralelos
    produtores = []
    for i in range(2):
        p = multiprocessing.Process(target=produtor, args=(queue, i, 5))
        produtores.append(p)
        p.start()
    
    # Criando consumidores paralelos
    consumidores = []
    for i in range(2):
        c = multiprocessing.Process(target=consumidor, args=(queue, i))
        consumidores.append(c)
        c.start()
    
    # Aguardar todos terminarem
    for p in produtores:
        p.join()
    for c in consumidores:
        c.join()
    
    print("Todos os processos finalizaram")
    
    # Exemplo com memória compartilhada (Array)
    print("\n=== Memória Compartilhada com Array ===")
    from multiprocessing import Array, Lock
    
    arr = Array('i', [0] * 10)  # Array de 10 inteiros
    lock = Lock()
    
    def atualizar_array(indice, valor):
        with lock:
            arr[indice] = valor
    
    processos = []
    for i in range(10):
        p = multiprocessing.Process(target=atualizar_array, args=(i, i * 10))
        processos.append(p)
        p.start()
    
    for p in processos:
        p.join()
    
    print(f"Array final: {list(arr)}")

import multiprocessing

import time

import random

def produtor(queue, id_produtor, n_itens):

"""Produz itens e coloca na fila."""

for i in range(n_itens):

item = f"Produtor-{id_produtor}-Item-{i}"

queue.put(item)

print(f"[Produtor {id_produtor}] Produziu: {item}")

time.sleep(random.uniform(0.1, 0.3))

queue.put(None) # Sinal de fim para este produtor

def consumidor(queue, id_consumidor):

"""Consome itens da fila até receber Nones suficientes."""

n_finalizados = 0

while True:

item = queue.get()

if item is None:

n_finalizados += 1

if n_finalizados == 2: # 2 produtores

break

continue

print(f"[Consumidor {id_consumidor}] Consumiu: {item}")

# Simula processamento do item

time.sleep(random.uniform(0.2, 0.5))

if __name__ == "__main__":

queue = multiprocessing.Queue()

# Criando produtores paralelos

produtores = []

for i in range(2):

p = multiprocessing.Process(target=produtor, args=(queue, i, 5))

produtores.append(p)

p.start()

# Criando consumidores paralelos

consumidores = []

for i in range(2):

c = multiprocessing.Process(target=consumidor, args=(queue, i))

consumidores.append(c)

c.start()

# Aguardar todos terminarem

for p in produtores:

p.join()

for c in consumidores:

c.join()

print("Todos os processos finalizaram")

# Exemplo com memória compartilhada (Array)

print("\n=== Memória Compartilhada com Array ===")

from multiprocessing import Array, Lock

arr = Array('i', [0] * 10) # Array de 10 inteiros

lock = Lock()

def atualizar_array(indice, valor):

with lock:

arr[indice] = valor

processos = []

for i in range(10):

p = multiprocessing.Process(target=atualizar_array, args=(i, i * 10))

processos.append(p)

p.start()

for p in processos:

p.join()

print(f"Array final: {list(arr)}")

Queues são seguras e ideais para padrões produtor-consumidor. Arrays compartilhados são eficientes para dados numéricos. Sempre use locks ao modificar memória compartilhada.

Escalabilidade e lei de amdahl

O ganho com processos paralelos não é linear infinito. A lei de Amdahl descreve o limite teórico de aceleração. A fórmula é: \(S = \frac{1}{(1 – P) + \frac{P}{N}}\) Onde P é a fração paralelizável e N é o número de núcleos. Por exemplo, com 90% paralelizável e 8 núcleos, o ganho máximo é 4.7x. Portanto, nem todo código pode ser perfeitamente paralelizado. A voz passiva é aplicada: “partes sequenciais são executadas em apenas um núcleo”. Exemplo demonstrando a lei de Amdahl na prática:

import multiprocessing
import time

def trabalho_paralelizavel(n):
    """Trabalho que pode ser dividido entre processos."""
    total = 0
    for i in range(n):
        total += i * i
    return total

def trabalho_sequencial(n):
    """Trabalho que precisa rodar em um único núcleo."""
    total = 0
    for i in range(n):
        total += i ** 0.5
    return total

def executar_com_paralelismo(n_tarefas, tamanho_par, tamanho_seq):
    """Executa parte paralela + parte sequencial."""
    # Parte sequencial (um núcleo)
    resultado_seq = trabalho_sequencial(tamanho_seq)
    
    # Parte paralela (múltiplos núcleos)
    with multiprocessing.Pool(processes=n_tarefas) as pool:
        resultados_par = pool.map(trabalho_paralelizavel, [tamanho_par] * n_tarefas)
    
    return resultado_seq, sum(resultados_par)

if __name__ == "__main__":
    n_nucleos = multiprocessing.cpu_count()
    print(f"Máquina com {n_nucleos} núcleos\n")
    
    # Experimento 1: 90% paralelizável
    tamanho_par = 40_000_000
    tamanho_seq = int(tamanho_par * 0.111)  # ~10% sequencial
    
    inicio = time.time()
    executar_com_paralelismo(1, tamanho_par, tamanho_seq)
    tempo_1_nucleo = time.time() - inicio
    
    inicio = time.time()
    executar_com_paralelismo(n_nucleos, tamanho_par, tamanho_seq)
    tempo_n_nucleos = time.time() - inicio
    
    aceleracao = tempo_1_nucleo / tempo_n_nucleos
    print(f"90% paralelizável:")
    print(f"  1 núcleo: {tempo_1_nucleo:.2f}s")
    print(f"  {n_nucleos} núcleos: {tempo_n_nucleos:.2f}s")
    print(f"  Aceleração real: {aceleracao:.2f}x")
    
    # Experimento 2: 50% paralelizável
    tamanho_par = 20_000_000
    tamanho_seq = tamanho_par  # 50% cada
    
    inicio = time.time()
    executar_com_paralelismo(1, tamanho_par, tamanho_seq)
    tempo_1_nucleo = time.time() - inicio
    
    inicio = time.time()
    executar_com_paralelismo(n_nucleos, tamanho_par, tamanho_seq)
    tempo_n_nucleos = time.time() - inicio
    
    aceleracao = tempo_1_nucleo / tempo_n_nucleos
    print(f"\n50% paralelizável:")
    print(f"  1 núcleo: {tempo_1_nucleo:.2f}s")
    print(f"  {n_nucleos} núcleos: {tempo_n_nucleos:.2f}s")
    print(f"  Aceleração real: {aceleracao:.2f}x")

import multiprocessing

import time

def trabalho_paralelizavel(n):

"""Trabalho que pode ser dividido entre processos."""

total = 0

for i in range(n):

total += i * i

return total

def trabalho_sequencial(n):

"""Trabalho que precisa rodar em um único núcleo."""

total = 0

for i in range(n):

total += i ** 0.5

return total

def executar_com_paralelismo(n_tarefas, tamanho_par, tamanho_seq):

"""Executa parte paralela + parte sequencial."""

# Parte sequencial (um núcleo)

resultado_seq = trabalho_sequencial(tamanho_seq)

# Parte paralela (múltiplos núcleos)

with multiprocessing.Pool(processes=n_tarefas) as pool:

resultados_par = pool.map(trabalho_paralelizavel, [tamanho_par] * n_tarefas)

return resultado_seq, sum(resultados_par)

if __name__ == "__main__":

n_nucleos = multiprocessing.cpu_count()

print(f"Máquina com {n_nucleos} núcleos\n")

# Experimento 1: 90% paralelizável

tamanho_par = 40_000_000

tamanho_seq = int(tamanho_par * 0.111) # ~10% sequencial

inicio = time.time()

executar_com_paralelismo(1, tamanho_par, tamanho_seq)

tempo_1_nucleo = time.time() - inicio

inicio = time.time()

executar_com_paralelismo(n_nucleos, tamanho_par, tamanho_seq)

tempo_n_nucleos = time.time() - inicio

aceleracao = tempo_1_nucleo / tempo_n_nucleos

print(f"90% paralelizável:")

print(f" 1 núcleo: {tempo_1_nucleo:.2f}s")

print(f" {n_nucleos} núcleos: {tempo_n_nucleos:.2f}s")

print(f" Aceleração real: {aceleracao:.2f}x")

# Experimento 2: 50% paralelizável

tamanho_par = 20_000_000

tamanho_seq = tamanho_par # 50% cada

inicio = time.time()

executar_com_paralelismo(1, tamanho_par, tamanho_seq)

tempo_1_nucleo = time.time() - inicio

inicio = time.time()

executar_com_paralelismo(n_nucleos, tamanho_par, tamanho_seq)

tempo_n_nucleos = time.time() - inicio

aceleracao = tempo_1_nucleo / tempo_n_nucleos

print(f"\n50% paralelizável:")

print(f" 1 núcleo: {tempo_1_nucleo:.2f}s")

print(f" {n_nucleos} núcleos: {tempo_n_nucleos:.2f}s")

print(f" Aceleração real: {aceleracao:.2f}x")

Observe que quanto maior a fração sequencial, menor o ganho. Portanto, identifique e otimize os gargalos sequenciais primeiro. Processos paralelos são ferramentas poderosas, mas não mágicas. Use-os com sabedoria e meça sempre o ganho real. Para muitos problemas, o paralelismo verdadeiro transforma horas em minutos. Experimente e veja seu código voar em múltiplos núcleos.