antonino, Autor em Á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.

Ray Computação Distribuída

22/05/202622/04/2026 Por antonino

Ray é um framework para computação distribuída em Python. Ele transforma funções e classes em tarefas remotas facilmente. Primeiramente, Ray escala de um laptop para um cluster gigante. Por exemplo, você pode paralelizar loops sem modificar muito código. Além disso, Ray é excelente para machine learning e reinforcement learning. Assim, cientistas de dados ganham produtividade imensa. Consequentemente, projetos que levavam dias agora levam horas. Portanto, a adoção de Ray cresce rapidamente na indústria. Quando utilizar Ray? Em computação paralela de larga escala. Também em treinamento de modelos, otimização de hiperparâmetros e simulações. Da mesma forma, qualquer tarefa paralelizável se beneficia. Por outro lado, para tarefas sequenciais simples, Ray é desnecessário. Ray oferece atores (estado compartilhado) e tarefas sem estado. Então, vamos explorar instalação, conceitos e exemplos práticos. Três subtítulos guiarão você pelo universo do Ray. Portanto, ao final, você paralelizará código como nunca antes.

Instalação e primeiros passos com ray

Instale Ray com pip: pip install ray. Depois, importe ray e inicie com ray.init(). O decorador @ray.remote torna uma função distribuída. Chamá-la com .remote() cria uma tarefa assíncrona. Use ray.get() para obter os resultados. Quando usar tarefas Ray? Em operações independentes e paralelizáveis. Por exemplo, processar imagens ou calcular estatísticas. Além disso, tarefas Ray são ideais para map-reduce simples. Exemplo básico de Ray:

import ray
import time

# Inicializa o Ray (local, sem cluster)
ray.init(ignore_reinit_error=True)

@ray.remote
def tarefa_pesada(n):
    """Função que será executada remotamente."""
    time.sleep(1)  # Simula trabalho pesado
    return n * n

# Versão sequencial
print("=== Execução sequencial ===")
inicio_seq = time.time()
resultados_seq = [tarefa_pesada(i) for i in range(10)]
tempo_seq = time.time() - inicio_seq
print(f"Tempo sequencial: {tempo_seq:.2f}s")

# Versão com Ray (paralela)
print("\n=== Execução com Ray ===")
inicio_ray = time.time()
# Cria tarefas remotas (não bloqueia)
futures = [tarefa_pesada.remote(i) for i in range(10)]
# Aguarda os resultados
resultados_ray = ray.get(futures)
tempo_ray = time.time() - inicio_ray
print(f"Tempo com Ray: {tempo_ray:.2f}s")
print(f"Aceleração: {tempo_seq / tempo_ray:.2f}x")

# Verificando resultados
print(f"Resultados: {resultados_ray[:5]}...")

# Fecha o Ray
ray.shutdown()

import ray

import time

# Inicializa o Ray (local, sem cluster)

ray.init(ignore_reinit_error=True)

@ray.remote

def tarefa_pesada(n):

"""Função que será executada remotamente."""

time.sleep(1) # Simula trabalho pesado

return n * n

# Versão sequencial

print("=== Execução sequencial ===")

inicio_seq = time.time()

resultados_seq = [tarefa_pesada(i) for i in range(10)]

tempo_seq = time.time() - inicio_seq

print(f"Tempo sequencial: {tempo_seq:.2f}s")

# Versão com Ray (paralela)

print("\n=== Execução com Ray ===")

inicio_ray = time.time()

# Cria tarefas remotas (não bloqueia)

futures = [tarefa_pesada.remote(i) for i in range(10)]

# Aguarda os resultados

resultados_ray = ray.get(futures)

tempo_ray = time.time() - inicio_ray

print(f"Tempo com Ray: {tempo_ray:.2f}s")

print(f"Aceleração: {tempo_seq / tempo_ray:.2f}x")

# Verificando resultados

print(f"Resultados: {resultados_ray[:5]}...")

# Fecha o Ray

ray.shutdown()

Ray executa tarefas em paralelo usando todos os núcleos. O ganho é próximo ao número de CPUs disponíveis. Portanto, para tarefas independentes, Ray é imbatível.

Atores: estado compartilhado entre tarefas

Atores são objetos Ray que mantêm estado entre chamadas. Use @ray.remote em uma classe para criar um ator. Instancie com Classe.remote() e chame métodos com .metodo.remote(). Quando usar atores? Em cenários com estado compartilhado. Por exemplo, contadores, acumuladores ou cache distribuído. Além disso, atores são úteis para simulações ou parâmetros compartilhados. Assim, você mantém consistência sem locks complexos. Atores mantêm estado consistente mesmo com chamadas concorrentes. Então, para sistemas com estado, prefira atores a tarefas puras. Exemplo de ator contador distribuído:

import ray
import time

ray.init(ignore_reinit_error=True)

@ray.remote
class ContadorDistribuido:
    """Ator que mantém um contador compartilhado."""
    
    def __init__(self):
        self.valor = 0
    
    def incrementar(self, quantidade=1):
        self.valor += quantidade
        return self.valor
    
    def obter_valor(self):
        return self.valor
    
    def resetar(self):
        self.valor = 0
        return self.valor

# Criando um ator (instância remota)
contador = ContadorDistribuido.remote()

# Chamando métodos do ator remotamente
print("=== Ator Contador Distribuído ===")
futures = []
for i in range(10):
    # Incrementa de forma distribuída
    futures.append(contador.incrementar.remote(1))

# Aguarda resultados parciais
resultados = ray.get(futures)
print(f"Incrementos: {resultados}")

# Obtém valor final
valor_final = ray.get(contador.obter_valor.remote())
print(f"Valor final do contador: {valor_final}")

# Exemplo com múltiplos atores independentes
print("\n=== Múltiplos Atores ===")
atores = [ContadorDistribuido.remote() for _ in range(5)]

# Incrementa cada ator com diferentes valores
for i, ator in enumerate(atores):
    ator.incrementar.remote(i + 1)

# Coleta resultados
valores = ray.get([ator.obter_valor.remote() for ator in atores])
print(f"Valores dos atores: {valores}")

# Exemplo com ator que acumula dados
@ray.remote
class Acumulador:
    def __init__(self):
        self.dados = []
    
    def adicionar(self, item):
        self.dados.append(item)
        return len(self.dados)
    
    def obter_todos(self):
        return self.dados
    
    def media(self):
        if not self.dados:
            return 0
        return sum(self.dados) / len(self.dados)

acumulador = Acumulador.remote()
for i in range(20):
    acumulador.adicionar.remote(i * 2)

# Obtém resultados
tamanho = ray.get(acumulador.adicionar.remote(100))
media = ray.get(acumulador.media.remote())
todos = ray.get(acumulador.obter_todos.remote())

print(f"\n=== Acumulador ===")
print(f"Total de itens: {tamanho}")
print(f"Média: {media:.2f}")
print(f"Primeiros 5 itens: {todos[:5]}...")

ray.shutdown()

import ray

import time

ray.init(ignore_reinit_error=True)

@ray.remote

class ContadorDistribuido:

"""Ator que mantém um contador compartilhado."""

def __init__(self):

self.valor = 0

def incrementar(self, quantidade=1):

self.valor += quantidade

return self.valor

def obter_valor(self):

return self.valor

def resetar(self):

self.valor = 0

return self.valor

# Criando um ator (instância remota)

contador = ContadorDistribuido.remote()

# Chamando métodos do ator remotamente

print("=== Ator Contador Distribuído ===")

futures = []

for i in range(10):

# Incrementa de forma distribuída

futures.append(contador.incrementar.remote(1))

# Aguarda resultados parciais

resultados = ray.get(futures)

print(f"Incrementos: {resultados}")

# Obtém valor final

valor_final = ray.get(contador.obter_valor.remote())

print(f"Valor final do contador: {valor_final}")

# Exemplo com múltiplos atores independentes

print("\n=== Múltiplos Atores ===")

atores = [ContadorDistribuido.remote() for _ in range(5)]

# Incrementa cada ator com diferentes valores

for i, ator in enumerate(atores):

ator.incrementar.remote(i + 1)

# Coleta resultados

valores = ray.get([ator.obter_valor.remote() for ator in atores])

print(f"Valores dos atores: {valores}")

# Exemplo com ator que acumula dados

@ray.remote

class Acumulador:

def __init__(self):

self.dados = []

def adicionar(self, item):

self.dados.append(item)

return len(self.dados)

def obter_todos(self):

return self.dados

def media(self):

if not self.dados:

return 0

return sum(self.dados) / len(self.dados)

acumulador = Acumulador.remote()

for i in range(20):

acumulador.adicionar.remote(i * 2)

# Obtém resultados

tamanho = ray.get(acumulador.adicionar.remote(100))

media = ray.get(acumulador.media.remote())

todos = ray.get(acumulador.obter_todos.remote())

print(f"\n=== Acumulador ===")

print(f"Total de itens: {tamanho}")

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

print(f"Primeiros 5 itens: {todos[:5]}...")

ray.shutdown()

Atores mantêm estado entre chamadas remotas. Isso é fundamental para muitos algoritmos distribuídos. Assim, você implementa sistemas complexos com facilidade.

Ray para machine learning e tunning de hiperparâmetros

Ray tem bibliotecas especializadas como Ray Tune e Ray Train. Ray Tune otimiza hiperparâmetros de modelos de ML. Ray Train distribui treinamento de modelos grandes. Quando usar essas ferramentas? Em projetos de ML profissional. Também em otimização de modelos com muitos parâmetros. Por exemplo, redes neurais ou algoritmos de ensemble. Além disso, Ray Serve serve modelos em produção. Exemplo de Ray Tune para otimização:

import ray
from ray import tune
from ray.tune.schedulers import ASHAScheduler
import numpy as np
import time

# Inicializa Ray
ray.init(ignore_reinit_error=True)

def funcao_objetivo(config):
    """Função a ser otimizada (exemplo: SVM com parâmetros)."""
    # Simula treinamento de modelo
    x = config["x"]
    y = config["y"]
    z = config["z"]
    
    # Função de custo simulada
    custo = (x - 1)**2 + (y - 2)**2 + (z - 3)**2
    time.sleep(0.1)  # Simula tempo de treinamento
    
    # Reporta métricas para o Tune
    tune.report(loss=custo, accuracy=1.0 / (1.0 + custo))

def demonstrar_ray_tune():
    """Demonstra otimização de hiperparâmetros com Ray Tune."""
    print("=== Ray Tune - Otimização de Hiperparâmetros ===\n")
    
    # Espaço de busca
    espaco_busca = {
        "x": tune.uniform(0, 2),
        "y": tune.uniform(1, 3),
        "z": tune.uniform(2, 4),
    }
    
    # Executa otimização
    analise = tune.run(
        funcao_objetivo,
        config=espaco_busca,
        num_samples=20,  # Número de combinações
        metric="loss",
        mode="min",
        progress_bar=True
    )
    
    # Melhores parâmetros encontrados
    melhores_config = analise.get_best_config(metric="loss", mode="min")
    melhor_loss = analise.get_best_trial(metric="loss", mode="min").last_result["loss"]
    
    print(f"\nMelhores parâmetros: x={melhores_config['x']:.3f}, "
          f"y={melhores_config['y']:.3f}, z={melhores_config['z']:.3f}")
    print(f"Melhor loss: {melhor_loss:.6f}")
    
    return melhores_config

# Exemplo com Ray para processamento paralelo de dados
@ray.remote
def processar_lote(dados_lote):
    """Processa um lote de dados em paralelo."""
    resultado = [x * 2 for x in dados_lote]
    time.sleep(0.01)  # Simula processamento
    return sum(resultado)

def processamento_paralelo_dados():
    """Divide dados em lotes e processa em paralelo."""
    dados = list(range(1000))
    num_lotes = 8
    tamanho_lote = len(dados) // num_lotes
    
    lotes = []
    for i in range(num_lotes):
        inicio = i * tamanho_lote
        fim = inicio + tamanho_lote if i < num_lotes - 1 else len(dados)
        lotes.append(dados[inicio:fim])
    
    print("\n=== Processamento Paralelo de Dados ===")
    inicio = time.time()
    
    # Processa lotes em paralelo
    futures = [processar_lote.remote(lote) for lote in lotes]
    resultados = ray.get(futures)
    total = sum(resultados)
    
    tempo = time.time() - inicio
    print(f"Processados {len(dados)} itens em {tempo:.3f}s")
    print(f"Total: {total}")
    
    # Comparação sequencial
    inicio_seq = time.time()
    total_seq = sum(x * 2 for x in dados)
    tempo_seq = time.time() - inicio_seq
    print(f"Tempo sequencial: {tempo_seq:.3f}s")
    print(f"Aceleração: {tempo_seq / tempo:.2f}x")

# Exemplo com Ray para simulação distribuída
@ray.remote
def simulacao(parametros):
    """Simula um sistema com parâmetros dados."""
    # Simulação complexa
    resultado = np.sin(parametros[0]) * np.cos(parametros[1])
    return resultado

def simulacoes_distribuidas():
    """Executa múltiplas simulações em paralelo."""
    print("\n=== Simulações Distribuídas ===")
    parametros_lista = [(i * 0.1, j * 0.1) for i in range(10) for j in range(10)]
    
    inicio = time.time()
    futures = [simulacao.remote(p) for p in parametros_lista]
    resultados = ray.get(futures)
    tempo = time.time() - inicio
    
    print(f"Executadas {len(parametros_lista)} simulações em {tempo:.2f}s")
    print(f"Média dos resultados: {np.mean(resultados):.4f}")

if __name__ == "__main__":
    # Demonstrações
    melhores = demonstrar_ray_tune()
    processamento_paralelo_dados()
    simulacoes_distribuidas()
    
    print("\n=== Resumo do Ray ===")
    print("Ray é ideal para:")
    print("  - Paralelização de loops pesados")
    print("  - Atores com estado compartilhado")
    print("  - Otimização de hiperparâmetros")
    print("  - Treinamento distribuído de ML")
    
    ray.shutdown()

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

import ray

from ray import tune

from ray.tune.schedulers import ASHAScheduler

import numpy as np

import time

# Inicializa Ray

ray.init(ignore_reinit_error=True)

def funcao_objetivo(config):

"""Função a ser otimizada (exemplo: SVM com parâmetros)."""

# Simula treinamento de modelo

x = config["x"]

y = config["y"]

z = config["z"]

# Função de custo simulada

custo = (x - 1)**2 + (y - 2)**2 + (z - 3)**2

time.sleep(0.1) # Simula tempo de treinamento

# Reporta métricas para o Tune

tune.report(loss=custo, accuracy=1.0 / (1.0 + custo))

def demonstrar_ray_tune():

"""Demonstra otimização de hiperparâmetros com Ray Tune."""

print("=== Ray Tune - Otimização de Hiperparâmetros ===\n")

# Espaço de busca

espaco_busca = {

"x": tune.uniform(0, 2),

"y": tune.uniform(1, 3),

"z": tune.uniform(2, 4),

}

# Executa otimização

analise = tune.run(

funcao_objetivo,

config=espaco_busca,

num_samples=20, # Número de combinações

metric="loss",

mode="min",

progress_bar=True

)

# Melhores parâmetros encontrados

melhores_config = analise.get_best_config(metric="loss", mode="min")

melhor_loss = analise.get_best_trial(metric="loss", mode="min").last_result["loss"]

print(f"\nMelhores parâmetros: x={melhores_config['x']:.3f}, "

f"y={melhores_config['y']:.3f}, z={melhores_config['z']:.3f}")

print(f"Melhor loss: {melhor_loss:.6f}")

return melhores_config

# Exemplo com Ray para processamento paralelo de dados

@ray.remote

def processar_lote(dados_lote):

"""Processa um lote de dados em paralelo."""

resultado = [x * 2 for x in dados_lote]

time.sleep(0.01) # Simula processamento

return sum(resultado)

def processamento_paralelo_dados():

"""Divide dados em lotes e processa em paralelo."""

dados = list(range(1000))

num_lotes = 8

tamanho_lote = len(dados) // num_lotes

lotes = []

for i in range(num_lotes):

inicio = i * tamanho_lote

fim = inicio + tamanho_lote if i < num_lotes - 1 else len(dados)

lotes.append(dados[inicio:fim])

print("\n=== Processamento Paralelo de Dados ===")

inicio = time.time()

# Processa lotes em paralelo

futures = [processar_lote.remote(lote) for lote in lotes]

resultados = ray.get(futures)

total = sum(resultados)

tempo = time.time() - inicio

print(f"Processados {len(dados)} itens em {tempo:.3f}s")

print(f"Total: {total}")

# Comparação sequencial

inicio_seq = time.time()

total_seq = sum(x * 2 for x in dados)

tempo_seq = time.time() - inicio_seq

print(f"Tempo sequencial: {tempo_seq:.3f}s")

print(f"Aceleração: {tempo_seq / tempo:.2f}x")

# Exemplo com Ray para simulação distribuída

@ray.remote

def simulacao(parametros):

"""Simula um sistema com parâmetros dados."""

# Simulação complexa

resultado = np.sin(parametros[0]) * np.cos(parametros[1])

return resultado

def simulacoes_distribuidas():

"""Executa múltiplas simulações em paralelo."""

print("\n=== Simulações Distribuídas ===")

parametros_lista = [(i * 0.1, j * 0.1) for i in range(10) for j in range(10)]

inicio = time.time()

futures = [simulacao.remote(p) for p in parametros_lista]

resultados = ray.get(futures)

tempo = time.time() - inicio

print(f"Executadas {len(parametros_lista)} simulações em {tempo:.2f}s")

print(f"Média dos resultados: {np.mean(resultados):.4f}")

if __name__ == "__main__":

# Demonstrações

melhores = demonstrar_ray_tune()

processamento_paralelo_dados()

simulacoes_distribuidas()

print("\n=== Resumo do Ray ===")

print("Ray é ideal para:")

print(" - Paralelização de loops pesados")

print(" - Atores com estado compartilhado")

print(" - Otimização de hiperparâmetros")

print(" - Treinamento distribuído de ML")

ray.shutdown()

Ray Tune busca automaticamente os melhores parâmetros. A fórmula do ganho em paralelismo é clara: \(S = \frac{T_{\text{seq}}}{T_{\text{ray}}} \approx N_{\text{workers}} \times (1 – O)\) Onde O é o overhead de comunicação (geralmente pequeno). Portanto, Ray é a ferramenta moderna para computação distribuída. Comece com tarefas simples e evolua para atores complexos. Finalmente, seu código escalará de um laptop para um cluster inteiro.