Apresentando o suporte do Apache Spark Join em sessões interativas do AWS Glue

Quando criamos sessões interativas do AWS Glue, nosso objetivo period tornar o AWS Glue tão interativo quanto executar Python native em um pocket book. Na maioria das vezes, tivemos sucesso. Com um pacote Python simples e um pocket book Jupyter, você pode executar remotamente no back-end do AWS Glue efêmero Spark. A abordagem baseada em Livy estava à frente de seu tempo, mas tinha limitações de seu protocolo baseado em REST. A execução do PySpark native desbloqueou recursos poderosos do ambiente de desenvolvimento integrado (IDE), como depuração e linting, para que seu ambiente pudesse entender o código e ajudá-lo a desenvolver aplicativos Spark mais rapidamente. Os clientes frequentemente dividiam seu trabalho de desenvolvimento. Eles usaram Spark native (ou contêineres Docker) para desenvolver em um IDE com uma pequena quantidade de dados e, em seguida, mudaram para sessões interativas do AWS Glue para validar o dimensionamento e o ajuste em relação ao conjunto de dados completo.

Com os lançamentos modernos do PySpark veio um novo protocolo: Apache Spark Join. O Spark Join preenche a lacuna entre esses dois mundos: você desenvolve em Python native, mas executa no AWS Glue com base em dados reais. Hoje, as sessões interativas do AWS Glue oferecem suporte nativo ao Spark Join. Você pode se conectar de qualquer ambiente que suporte o PySpark distant() API, incluindo VS Code, PyCharm, notebooks Amazon SageMaker Unified Studio e aplicativos Python independentes. Você não precisa instalar kernels especializados ou gerenciar a infraestrutura de cluster.

O que o Spark Join muda

O Spark Join, introduzido no Spark 3.4, desacopla o cliente Spark do servidor por meio de um protocolo gRPC leve. Em vez de executar seu programa de driver no cluster, seu IDE se comunica com um servidor Spark remoto por meio de uma camada de cliente fino. Essa arquitetura libera a principal melhoria do fluxo de trabalho: você desenvolve localmente e executa remotamente.

Arquitetura Spark Join — skinny shopper com todo o poder do Apache Spark

Com o suporte do Spark Join em sessões interativas do AWS Glue, você obtém:

• Liberdade de IDE – Use VS Code, PyCharm, JupyterLab ou qualquer ambiente Python. Nenhuma instalação do kernel é necessária.
• Acesso programático – Crie o Spark em seus aplicativos Python e scripts de automação com um padrão SparkSession.builder.distant() chamar.
• Execução sem servidor – O AWS Glue provisiona e gerencia o cluster Spark. Você paga apenas pelas unidades de processamento de dados (DPUs) consumidas enquanto sua sessão estiver ativa.
• Monitoramento Spark Join – A UI do Spark Dwell agora inclui uma guia Join dedicada que mostra sessões e operações ativas do Spark Join junto com as visualizações existentes de Jobs, Estágios e Executores.

Primeiros passos com o SageMaker Unified Studio

O Amazon SageMaker Unified Studio fornece o caminho mais direto para o Spark Join no AWS Glue. O ambiente de pocket book lida com a criação de sessão, recuperação de endpoint e atualização de token automaticamente, portanto, nenhum padrão de conexão é necessário.

Pré-requisito: Você precisa de um projeto do Amazon SageMaker Unified Studio para usar esse fluxo de trabalho. Se você não tiver um, primeiro crie um projeto em seu domínio do SageMaker Unified Studio.

Para se conectar a uma sessão do AWS Glue Spark Join:

1. Faça login no SageMaker Unified Studio, escolha seu projeto e crie ou abra um Pocket book.

Um bloco de anotações aberto no SageMaker Unified Studio

2. Escolha o ícone de computação na barra de ferramentas esquerda para abrir o painel Ambiente de computação. Expanda o Fagulha seção.

O painel Ambiente de computação com a lista suspensa Spark

3. Selecione uma conexão Glue Spark. Dependendo da configuração do seu domínio SageMaker, você verá default.spark ou conexões nomeadas, como undertaking.spark.compatibility. Selecione a conexão de cola (Spark) apropriada e escolha Aplicar.

Conectado ao Glue Spark Join — em execução spark.model retorna ‘3.5.6-amzn-1’

Depois de fazer sua seleção, você estará conectado. O spark objeto de sessão está disponível nativamente. Nenhuma importação ou configuração é necessária. Comece a executar o PySpark imediatamente:

spark.sql("SHOW DATABASES").present()

A sessão é gerenciada em segundo plano, incluindo atualização automática de token.

Usando o SDK sagemaker_studio

O sagemaker-studio O pacote Python estende a experiência do Spark Join além dos notebooks SageMaker Unified Studio para IDEs locais, pipelines de integração contínua e entrega contínua (CI/CD) e qualquer ambiente Python. O sparkutils O módulo lida com a inicialização da sessão e a configuração da conexão em uma única chamada. Você obtém a mesma experiência simplificada do pocket book, em qualquer lugar em que executa o Python:

from sagemaker_studio import sparkutils

# Initialize a Glue Spark Join session utilizing your undertaking connection
spark = sparkutils.init(connection_name="default.spark")

# Run queries instantly
spark.sql("SHOW DATABASES").present()

Você também pode usar sparkutils.get_spark_options() para recuperar opções pré-configuradas de Java Database Connectivity (JDBC) para leitura e gravação em fontes de dados por meio de conexões de projeto. As fontes compatíveis incluem Amazon Redshift, Amazon Aurora e Amazon DocumentDB (com compatibilidade com MongoDB):

# Get connection choices for a Redshift connection in your undertaking
choices = sparkutils.get_spark_options("my_redshift_connection")

# Learn from Redshift through Spark Join
df = spark.learn.format("jdbc").choices(**choices).possibility("dbtable", "analytics.orders").load()
df.present()

No SageMaker Unified Studio, o sagemaker-studio SDK é nativo do ambiente. O spark sessão e sparkutils estão disponíveis sem instalação. Para uso IDE native, instale-o com pip set up sagemaker-studio e configure credenciais por meio de um perfil nomeado da AWS ou sessão boto3.

Como funciona

As sessões do Spark Join no AWS Glue usam um fluxo de trabalho de três etapas:

1. Crie uma sessão – Ligue para o CreateSession API com SessionType definido como SPARK_CONNECT. A sessão é provisionada em aproximadamente 30 segundos.
2. Recuperar o ponto remaining – Chamar GetSessionEndpoint para receber um sc:// URL do endpoint gRPC e um token de autenticação por tempo limitado.
3. Conecte-se com PySpark – Passe o endpoint e o token para SparkSession.builder.distant() e comece a executar operações do Spark.

Fluxo do protocolo Spark Join – API DataFrame traduzida para plano lógico, enviada through gRPC/protobuf, resultados transmitidos de volta through gRPC/Arrow

Conectando-se com a API de baixo nível

Alguns ambientes não possuem o sagemaker-studio SDK, como contêineres personalizados, funções do AWS Lambda ou conjuntos de ferramentas não Python. Nesses ambientes, ou se não estiver usando o SageMaker Unified Studio, você pode usar o AWS SDK (Boto3) para gerenciar sessões diretamente. O exemplo a seguir demonstra o fluxo de trabalho completo:

import time, boto3, urllib.parse
from pyspark.sql import SparkSession

glue = boto3.shopper("glue", region_name="us-east-1")

# 1. Create a Spark Join session
session_id = "my-spark-connect-session"
glue.create_session(
    Id=session_id,
    Function="arn:aws:iam::123456789012:position/GlueServiceRole",
    Command={"Title": "glueetl"},
    GlueVersion="5.1",
    SessionType="SPARK_CONNECT",
    DefaultArguments={"--enable-spark-live-ui": "true"},
)

# 2. Watch for the session to achieve READY
whereas True:
    standing = glue.get_session(Id=session_id)("Session")("Standing")
    if standing == "READY":
        break
    time.sleep(5)

# 3. Get the Spark Join endpoint
sc = glue.get_session_endpoint(SessionId=session_id)("SparkConnect")
endpoint_url = sc("Url")
auth_token = sc("AuthToken")

# 4. Join with PySpark
encoded_token = urllib.parse.quote(auth_token, secure="")
connection_string = f"{endpoint_url}:443/;use_ssl=true;x-aws-proxy-auth={encoded_token}"
spark = SparkSession.builder.distant(connection_string).getOrCreate()
spark.sql("SELECT 1 + 1 AS end result").present()

Monitoramento com Spark Dwell UI

Ao ativar a UI do Spark Dwell na criação da sessão, você obtém acesso a um painel em tempo actual que mostra:

• Empregos e etapas – Rastreie trabalhos ativos, concluídos e com falha com métricas em nível de estágio.
• Executores – Monitore o uso da memória, embaralhe os dados e a integridade do executor.
• SQL – Inspecione planos de consulta e detalhes de execução.
• Guia Conectar – Visualize sessões e operações ativas do Spark Join (específicas do Spark Join).

Acesse o painel através do GetDashboardUrl API ou diretamente do console do AWS Glue.

import boto3, webbrowser

glue = boto3.shopper("glue", region_name="us-east-1")
dashboard = glue.get_dashboard_url(
    ResourceId="my-spark-connect-session",
    ResourceType="SESSION",
)
webbrowser.open(dashboard("Url"))

No SageMaker Unified Studio, nenhuma chamada de API é necessária. Escolher Preparar na barra de standing do pocket book para abrir o popover de informações do kernel. A partir daí, abra o IU do Spark hyperlink para o painel ao vivo ou Registros do driver Spark para saída de log em tempo actual.

Imagem mostrando “Pronto” na barra de standing para acessar Spark UI e Driver Logs diretamente do pocket book

Atualização de token

Os tokens de autenticação expiram após 30 minutos. No SageMaker Unified Studio, isso é tratado automaticamente. Para uso programático, você pode usar um thread em segundo plano para manter a conexão ativa. O seguinte auxiliar se reconecta de forma transparente antes que o token expire:

import threading, time, boto3, urllib.parse
from pyspark.sql import SparkSession

class GlueSparkConnect:
    """Maintains a SparkSession with computerized token refresh."""

    def __init__(self, session_id, area="us-east-1", refresh_margin=300):
        self.session_id = session_id
        self.glue = boto3.shopper("glue", region_name=area)
        self.refresh_margin = refresh_margin  # seconds earlier than expiry to refresh
        self._lock = threading.Lock()
        self.spark = self._connect()
        self._start_refresh_loop()

    def _connect(self):
        sc = self.glue.get_session_endpoint(SessionId=self.session_id)("SparkConnect")
        encoded_token = urllib.parse.quote(sc("AuthToken"), secure="")
        remote_url = f"{sc('Url')}:443/;use_ssl=true;x-aws-proxy-auth={encoded_token}"
        self._token_expiry = sc("AuthTokenExpirationTime").timestamp()
        return SparkSession.builder.distant(remote_url).getOrCreate()

    def _start_refresh_loop(self):
        def _loop():
            whereas True:
                sleep_for = max(self._token_expiry - time.time() - self.refresh_margin, 30)
                time.sleep(sleep_for)
                with self._lock:
                    self.spark = self._connect()
        t = threading.Thread(goal=_loop, daemon=True)
        t.begin()

# Utilization
session = GlueSparkConnect("my-spark-connect-session")
session.spark.sql("SELECT 1 + 1 AS end result").present()

O thread em segundo plano fica suspenso até 5 minutos antes da expiração do token e depois se reconecta de forma transparente. Como o thread daemon termina quando o script termina, não há necessidade de limpeza.

Começando

Para começar a usar o Spark Join com sessões interativas do AWS Glue:

1. Use o AWS Glue versão 5.1 (Apache Spark 3.5.6).
2. Instale o PySpark 3.5.6 localmente: pip set up pyspark==3.5.6.
3. Conceda permissões de identidade do AWS Id and Entry Administration (IAM) para glue:CreateSession, glue:GetSessione glue:GetSessionEndpoint.
4. Crie uma sessão com --session-type SPARK_CONNECT e conecte-se a partir do seu ambiente preferido.

Observação sobre VPC: se você se conectar a sessões interativas do AWS Glue por meio de um endpoint de nuvem privada digital (VPC), adicione o novo endpoint do Spark Join (com.amazonaws.{area}.glue.periods) à configuração da sua VPC. Os endpoints VPC existentes do AWS Glue não cobrem o tráfego do Spark Join.

Para obter instruções detalhadas, consulte Conectar-se a uma sessão do Spark Join no Guia do desenvolvedor do AWS Glue.

Sobre os autores