Пишем высокопроизводительный rate-limiter на golang
О чём статья
Недавно я познакомился с golang и в качестве практики решил реализовать алгоритмы rate limiting'а из статьи c arxiv. В этой статье мы рассмотрим написание rate-limiter'а в виде отдельно стоящего сервиса и в виде middleware, используя in-memory кэш или Redis, добавим поддержку REST и gRPC, подключим Prometheus и поймём как добиться атомарности операций с минимальными задержками.
Кому полезно
Новичкам в golang или же людям, которые по какой-то причине также решили использовать собственные реализации rate-limiter'ов заместо уже существующих решений.
Предисловие
Указанная ранее статья покрывает 3 алгоритма rate-limiting'а:
Token bucket
Fixed window
Rolling window
Суть данных алгоритмов вы можете прочитать самостоятельно, я же сфокусируюсь на том, как написать данные алгоритмы таким образом, чтобы rate-limiter работал быстро и надёжно.
Конфигурация
Первое, с чего стоит начать, это конфигурация вашего limiter'а. В своей реализации я ставил следующие требования:
Возможность переключаться между алгоритмами
Поддержка in-memory кэша и Redis
Метрики в Prometheus
Поддержка REST и gRPC
Из этих требований вытекает следующий файл конфигурации:
store: "redis"
use_algo: "token_bucket"
metrics:
use_prometheus: true
port: 9091
algo_settings:
capacity: 10
rate: 0.1
backends:
in_memory:
default_ttl: 30 # seconds
eviction_time: 120 # seconds
shards_count: 256
redis:
addr: "redis:6379"
password: "m1re1di77sss"
db: 0
default_ttl: 1000sourcecraft.dev
Рассмотрим его подробнее:
store отвечает за используемое хранилище, может быть "redis" или же "in_memory"
use_algo отвечает за используемый алгоритм
metrics позволяет включать и отключать поддержку выгрузки и подсчёта метрик во время работы, а также за порт, на котором позже будет поднят мультиплексер для метрик
algo_settings для каждого алгоритма разные, отвечает за настройки, которые будут использоваться алгоритмом
backends - настройки каждого из хранилищ, в in_memory default_ttl отвечает за ttl ключей в хранилище, eviction_time отвечает за частоту очистки in-memory хранилища от протухших ключей, shard_count отвечает за то, на сколько частей in-memory кэш будет шардирован, число должно быть степенью 2, позже я объясню почему
Конфигурация для middleware выглядит похожим образом:
type InMemoryBackendSettings struct {
DefaultTtl int64 yaml:"default_ttl"
EvictionTime int64 yaml:"eviction_time"
ShardsCount uint32 yaml:"shards_count"
}
type RedisBackendSettings struct {
Addr string yaml:"addr"
Password string yaml:"password"
Db int yaml:"db"
DefaultTtl int64 yaml:"default_ttl"
}
type AlgoSettings struct {
Capacity *float64 yaml:"capacity"
Rate *float64 yaml:"rate"
WindowSize *int64 yaml:"window_size"
MaxRequests *int64 yaml:"max_requests"
}sourcecraft.dev
Бэкенды для limiter'а
Думаю, ни для кого не секрет, что Redis умеет выполнять Lua скрипты атомарно, что является идеальной функциональной возможностью для надёжного rate-limiting'а. В таком случае наш бэкенд limiter'а также должен уметь выполнять операции атомарно, отсюда вытекает интерфейс бэкенда:
type InMemoryStoreInterface interface {
Update(key string, newFn func() any, fn func(v any) error) error
Close() error
}sourcecraft.dev
Update принимает на вход ключ, по которому мы хотим принять решение, а также 2 колбэка:
newFn - функция создания нового объекта в хранилище, в случае если данный ключ в хранилище ещё не появлялся. Это позволяет избежать создания данного объекта в случае, если ключ уже содержится в хранилище. Конечно, можно было сначала проверять и только потом добавлять, но это добавляет лишней логики клиенту, использующему хранилище.
fn - собственно функция, содержащая логику limiting'а
In-memory кэш
Наше хранилище под капотом будет использовать мапу, во избежание race condition'ов мы будем использовать mutex. Но ставить mutex на всю мапу слишком дорого, так как другие клиенты всё ещё должны пользоваться хранилищем.
Отсюда вытекает необходимость в шардировании. Пока одна горутина выполняет логику по ключу x, мы всё ещё можем работать с ключом y:
type shard struct {
mu sync.Mutex
entries map[string]*cacheEntry
}
type cacheEntry struct {
entry any
lastAccess int64
}
type ShardedInMemoryStore struct {
shards []*shard
shardsCount uint32
expirationTime int64
ticker *time.Ticker
done chan any
once sync.Once
}
func NewCache(
expirationTime int64,
evictionInterval time.Duration,
shardCount uint32,
) *ShardedInMemoryStore {
ticker := time.NewTicker(evictionInterval)
cache := &ShardedInMemoryStore{
shards: make([]*shard, shardCount),
shardsCount: shardCount,
expirationTime: expirationTime,
done: make(chan any),
ticker: ticker,
}
for i := range shardCount {
cache.shards[i] = &shard{entries: make(map[string]*cacheEntry)}
}
go func() {
defer ticker.Stop()
for {
select {
case <-cache.done:
return
case <-ticker.C:
cache.handleEviction()
}
}
}()
return cache
}
sourcecraft.dev
Также удаление протухших ключей теперь не будет лочить всю мапу, а только один конкретный шард:
func (cache *ShardedInMemoryStore) handleEviction() {
currentTime := time.Now().Unix()
for _, shard := range cache.shards {
shard.mu.Lock()
for key, value := range shard.entries {
if currentTime-cache.expirationTime > value.lastAccess {
delete(shard.entries, key)
}
}
shard.mu.Unlock()
}
}sourcecraft.dev
А также ранее упомянутый Update:
func (cache *ShardedInMemoryStore) Update(key string, createFn func() any, fn func(v any) error) error {
shard := cache.getShard(key)
shard.mu.Lock()
defer shard.mu.Unlock()
now := time.Now().Unix()
value, present := shard.entries[key]
if !present {
value = &cacheEntry{entry: createFn(), lastAccess: now}
shard.entries[key] = value
}
value.lastAccess = now
return fn(value.entry)
}sourcecraft.dev
Именно здесь реализуется столь необходимая атомарность, мы лочим mutex на шарде и выполняем операцию, не блокируя другие шарды и не допуская race condition'ов.
Так почему количество шардов должно быть степенью двойки?
Для того, чтобы распределять клиентские ключи более-менее равномерно по шардам нужно использоват функцию хэширования, которая не только будет раскладывать ключи равномерно, но и будет быстрой. Под такую функцию подходит Fowler–Noll–Vo. А чтобы после этого быстро замапить хэш на шард можно воспользоваться трюком. Вкратце можно очень быстро искать остатки от деления на степень двойки просто используя формулу: hash & (shardsCount - 1):
func (cache *ShardedInMemoryStore) getShard(key string) *shard {
h := uint32(2166136261)
for i := 0; i < len(key); i++ {
h ^= uint32(key[i])
h *= 16777619
}
return cache.shards[h&(cache.shardsCount-1)]
}sourcecraft.dev
Redis кэш
С Redis всё проще:
package cache
import (
"context"
"github.com/redis/go-redis/v9"
)
type RedisStoreInterface interface {
RunLuaScript(ctx context.Context, script *redis.Script, keys []string, args ...any) (bool, int64, error)
DefaultTtl() int64
}sourcecraft.dev
Передаём контекст, скрипт и массив ключей, в данном случае массив ключей это требование библиотеки для работы с Redis в Golang. Массив всегда будет состоять из одного элемента.
DefaultTtl - это getter, который используется для token bucket алгоритма, чтобы удалением ключа при использовании Redis занимался сам Redis, а не приложение.
package cache
import (
"context"
"fmt"
"github.com/redis/go-redis/v9"
)
type RedisStore struct {
client *redis.Client
defaultTtl int64
}
func NewRedisStore(client *redis.Client, defaultTtl int64) *RedisStore {
return &RedisStore{client: client, defaultTtl: defaultTtl}
}
// RunLuaScript runs a Lua script that returns {allowed(0|1), retryAfter}.
func (r *RedisStore) RunLuaScript(
ctx context.Context,
script *redis.Script,
keys []string,
args ...any,
) (bool, int64, error) {
res, err := script.Run(ctx, r.client, keys, args...).Int64Slice()
if err != nil {
return false, 0, err
}
if len(res) != 2 {
return false, 0, fmt.Errorf("unexpected length of script output")
}
return res[0] == 1, res[1], nil
}
func (r *RedisStore) DefaultTtl() int64 {
return r.defaultTtl
}sourcecraft.dev
Используем созданные хранилища
Теперь рассмотрим функцию запуска limiter'а:
package main
import (
"context"
"fmt"
"log"
"net"
"net/http"
"os"
"os/signal"
"qwrttqr-rate-limiter/core/config"
"qwrttqr-rate-limiter/core/server/algos"
"qwrttqr-rate-limiter/core/server/facade"
"qwrttqr-rate-limiter/core/server/limiting"
"qwrttqr-rate-limiter/core/server/metrics"
ratelimiterv1 "qwrttqr-rate-limiter/gen/ratelimiter/v1"
"strconv"
"syscall"
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promhttp"
"google.golang.org/grpc"
)
func main() {
if err := run(); err != nil {
log.Fatal(err)
}
}
func run() error {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
cfg, err := config.ReadConfig()
if err != nil {
return fmt.Errorf("failed to read config: %w", err)
}
rl, err := limiting.NewRateLimiter(cfg)
if err != nil {
return fmt.Errorf("create rate limiter: %w", err)
}
defer rl.Close()
var limiter algos.Limiter = rl
errCh := make(chan error, 3)
shutdownMetrics := func(ctx context.Context) {}
if cfg.Metrics.UsePrometheus {
reg := prometheus.NewRegistry()
rec := metrics.NewPromRecorder(reg)
limiter = metrics.Instrument(limiter, rec)
metricsMux := http.NewServeMux()
metricsMux.Handle("GET /metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))
metricsSrv := &http.Server{Addr: ":" + strconv.Itoa(cfg.Metrics.Port), Handler: metricsMux, ReadHeaderTimeout: 5 * time.Second}
go func() { errCh <- metricsSrv.ListenAndServe() }()
log.Printf("metrics available on %s", metricsSrv.Addr)
shutdownMetrics = func(ctx context.Context) {
_ = metricsSrv.Shutdown(ctx)
}
}
limitingMux := http.NewServeMux()
limitingMux.HandleFunc("GET /health", facade.HealthHandler)
limitingMux.HandleFunc("POST /limit", facade.LimitHandler(limiter))
limitingSrv := &http.Server{Addr: ":8080", Handler: limitingMux, ReadHeaderTimeout: 5 * time.Second}
grpcSrv := grpc.NewServer()
ratelimiterv1.RegisterRateLimiterServer(grpcSrv, facade.NewGRPCServer(limiter))
tcpLis, err := net.Listen("tcp", ":8081")
if err != nil {
return fmt.Errorf("listen grpc: %w", err)
}
go func() { errCh <- limitingSrv.ListenAndServe() }()
log.Printf("listening http on %s", limitingSrv.Addr)
go func() { errCh <- grpcSrv.Serve(tcpLis) }()
log.Printf("listening grpc on %s", tcpLis.Addr())
select {
case err := <-errCh:
return err
case <-ctx.Done():
}
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
grpcDone := make(chan struct{})
go func() { grpcSrv.GracefulStop(); close(grpcDone) }()
select {
case <-grpcDone:
case <-shutdownCtx.Done():
grpcSrv.Stop()
}
shutdownMetrics(shutdownCtx)
return limitingSrv.Shutdown(shutdownCtx)
}
sourcecraft.dev
Давайте по порядку:
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()sourcecraft.dev
Создаём shutdown context, который срабатывает на нажатие Ctrl-C либо при получении syscall.SIGTERM (сигнал, посылаемый родительским процессом на завершение нашего процесса, например когда тушим докер-контейнер). defer stop() нужен для того, чтобы прекратить обрабатывать данные сигналы, сразу перед возвратом из run.
cfg, err := config.ReadConfig()
if err != nil {
return fmt.Errorf("failed to read config: %w", err)
}
rl, err := limiting.NewRateLimiter(cfg)
if err != nil {
return fmt.Errorf("create rate limiter: %w", err)
}
defer rl.Close()sourcecraft.dev
Далее читаем конфиг и создаём сам limiter, у limiter'а есть функция Close, нужная для закрытия соединения с Redis. в in-memory кэше она просто закрывает канал, что останавливает удаление ключей из кэша.
var limiter algos.Limiter = rl
errCh := make(chan error, 3)sourcecraft.dev
Следующие строки позволяют создать указатель с типом интерфейса на limiter:
package algos
import (
"context"
)
type Request struct {
Key string
RequiredTokens float64
}
type Decision struct {
Allowed bool
// RetryAfter is seconds
RetryAfter int64
}
type Limiter interface {
Allow(ctx context.Context, req Request) (Decision, error)
}sourcecraft.dev
Так как NewRateLimiter возвращает не интерфейс, а указатель на структуру, поддерживающую Close и Allow:
type RateLimiter struct {
algos.Limiter
io.Closer
}
func NewRateLimiter(cfg config.StandaloneServiceConfiguration) (_ *RateLimiter, err error) {
var (
mem cache.InMemoryStoreInterface
rdb *redis.Client
closer io.Closer
)
switch cfg.Store {
case "in_memory":
cacheInstance := cache.NewCache(
cfg.Backends.InMemory.DefaultTtl,
time.Duration(cfg.Backends.InMemory.EvictionTime)*time.Second,
cfg.Backends.InMemory.ShardsCount,
)
mem, closer = cacheInstance, cacheInstance
case "redis":
redisClient := redis.NewClient(&redis.Options{
Addr: cfg.Backends.Redis.Addr,
Password: cfg.Backends.Redis.Password,
DB: cfg.Backends.Redis.Db,
})
rdb, closer = redisClient, redisClient
default:
return nil, fmt.Errorf("unsupported store: %q", cfg.Store)
}
defer func() {
if err != nil {
_ = closer.Close()
}
}()
var limiter algos.Limiter
switch cfg.UsedAlgo {
case "token_bucket":
if err := algos.ValidateTokenBucketConfig(cfg.AlgoSettings); err != nil {
return nil, err
}
store, err := algos.NewTokenBucketStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl)
if err != nil {
return nil, err
}
limiter = &algos.TokenBucketLimiter{
Capacity: *cfg.AlgoSettings.Capacity,
Rate: *cfg.AlgoSettings.Rate,
Store: store,
Now: realClock,
}
case "fixed_window":
if err := algos.ValidateFixedWindowConfig(cfg.AlgoSettings); err != nil {
return nil, err
}
store, err := algos.NewFixedWindowStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl)
if err != nil {
return nil, err
}
limiter = &algos.FixedWindowLimiter{
MaxRequests: *cfg.AlgoSettings.MaxRequests,
WindowSize: *cfg.AlgoSettings.WindowSize,
Store: store,
Now: realClock,
}
case "rolling_window":
if err := algos.ValidateRollingWindowConfig(cfg.AlgoSettings); err != nil {
return nil, err
}
store, err := algos.NewRollingWindowStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl)
if err != nil {
return nil, err
}
limiter = &algos.RollingWindowLimiter{
MaxRequests: *cfg.AlgoSettings.MaxRequests,
WindowSize: *cfg.AlgoSettings.WindowSize,
Store: store,
Now: realClock,
}
default:
return nil, fmt.Errorf("unsupported rate limiting algorithm: %s", cfg.UsedAlgo)
}
return &RateLimiter{Limiter: limiter, Closer: closer}, nil
}sourcecraft.dev
Возвращаясь обратно в run:
shutdownMetrics := func(ctx context.Context) {}
if cfg.Metrics.UsePrometheus {
reg := prometheus.NewRegistry()
rec := metrics.NewPromRecorder(reg)
limiter = metrics.Instrument(limiter, rec)
metricsMux := http.NewServeMux()
metricsMux.Handle("GET /metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))
metricsSrv := &http.Server{Addr: ":" + strconv.Itoa(cfg.Metrics.Port), Handler: metricsMux, ReadHeaderTimeout: 5 * time.Second}
go func() { errCh <- metricsSrv.ListenAndServe() }()
log.Printf("metrics available on %s", metricsSrv.Addr)
shutdownMetrics = func(ctx context.Context) {
_ = metricsSrv.Shutdown(ctx)
}
}sourcecraft.dev
Мы помечали rl, как структуру, реализующую интерфейс как раз ради поддержки метрик здесь. О том, как метрики работают я расскажу позже, сейчас достаточно понимания того, что мы создаём специальный HTTP сервер, с которого Prometheus будет забирать метрики. Для подсчёта данных метрик мы используем Registry, а в функции instrument мы оборачиваем limiter в мидлварь, которая будет обновлять счётчики.
Функция shutdownMetrics обновляется в зависимости от того, включаем ли у себя метрики или нет, это нужно, чтобы сделать graceful shutdown данного сервера.
limitingMux := http.NewServeMux()
limitingMux.HandleFunc("GET /health", facade.HealthHandler)
limitingMux.HandleFunc("POST /limit", facade.LimitHandler(limiter))
limitingSrv := &http.Server{Addr: ":8080", Handler: limitingMux, ReadHeaderTimeout: 5 * time.Second}
grpcSrv := grpc.NewServer()
ratelimiterv1.RegisterRateLimiterServer(grpcSrv, facade.NewGRPCServer(limiter))
tcpLis, err := net.Listen("tcp", ":8081")
if err != nil {
return fmt.Errorf("listen grpc: %w", err)
}
go func() { errCh <- limitingSrv.ListenAndServe() }()
log.Printf("listening http on %s", limitingSrv.Addr)
go func() { errCh <- grpcSrv.Serve(tcpLis) }()
log.Printf("listening grpc on %s", tcpLis.Addr())
select {
case err := <-errCh:
return err
case <-ctx.Done():
}sourcecraft.dev
После этого создаём HTTP и gRPC сервера и ждём в select на ранее созданном NotifyContext, до момента пока не придёт сигнал на завершение процесса (os.Interrupt, syscall.SIGTERM) или ошибка.
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
grpcDone := make(chan struct{})
go func() { grpcSrv.GracefulStop(); close(grpcDone) }()
select {
case <-grpcDone:
case <-shutdownCtx.Done():
grpcSrv.Stop()
}
shutdownMetrics(shutdownCtx)
return limitingSrv.Shutdown(shutdownCtx)sourcecraft.dev
В случае, если к нам пришла ошибка или же сигнал на завершение процесса, создаём контекст с таймаутом и просим все наши сервера выключиться. В особенности это касается gRPC сервера, он самостоятельно поддерживает GracefulStop, но если он не успеет сделать это за 10 секунд, мы выключим его принудительно.
Используем ранее созданные бэкенды
Помните, что мы создали Redis и InMemory бэкенды для наших алгоритмов, теперь можно посмотреть, как мы пользуемся ими, рассмотрим пример с Rolling Window:
type RollingWindowStore interface {
CheckAndIncrement(ctx context.Context, key string, windowSize, maxRequests, now int64) (bool, int64, error)
}
type RollingWindowLimiter struct {
WindowSize int64
MaxRequests int64
Store RollingWindowStore
Now func() int64
}
func (rwl *RollingWindowLimiter) Configure() {
if rwl.Now == nil {
rwl.Now = func() int64 {
return time.Now().Unix()
}
}
}
func ValidateRollingWindowConfig(cfg config.AlgoSettings) error {
if *cfg.MaxRequests <= 0 {
return fmt.Errorf("the max requests param should be strictly greater then 0")
}
return CheckRequiredFields(cfg, []string{"window_size", "max_requests"})
}
type InMemoryRollingWindowStore struct {
Backend cache.InMemoryStoreInterface
}
type RedisRollingWindowStore struct {
Backend cache.RedisStoreInterface
}
func NewRollingWindowStore(store string, cacheInstance cache.InMemoryStoreInterface, redisClient *redis.Client, redisDefaultKeyTTL int64) (RollingWindowStore, error) {
switch store {
case "in_memory":
return &InMemoryRollingWindowStore{Backend: cacheInstance}, nil
case "redis":
return &RedisRollingWindowStore{Backend: cache.NewRedisStore(redisClient, redisDefaultKeyTTL)}, nil
default:
return nil, fmt.Errorf("unsupported store: %s", store)
}
}sourcecraft.dev
Ещё один интерфейс(RollingWindowStore), нужен для генерализации работы с бэкендом, таким образом, сам алгоритм будет просто вызывать функцию CheckAndIncrement, передавая параметры, а бэкенд будет выполнять её в атомарном варианте.
func (rwl *RollingWindowLimiter) Configure() {
if rwl.Now == nil {
rwl.Now = func() int64 {
return time.Now().Unix()
}
}
}sourcecraft.dev
Данная функция нужна для тестов, так как в тестах придётся мокать функцию получения текущего времени.
type RollingWindowState struct {
Timestamps []int64
}
func (rws *RollingWindowState) check(
windowSize,
maxRequests,
now int64,
) (bool, int64) {
windowStart := now - windowSize
i := 0
for ; i < len(rws.Timestamps); i++ {
if rws.Timestamps[i] > windowStart {
break
}
}
if i > 0 {
copy(rws.Timestamps, rws.Timestamps[i:])
rws.Timestamps = rws.Timestamps[:len(rws.Timestamps)-i]
}
if int64(len(rws.Timestamps)) < maxRequests {
rws.Timestamps = append(rws.Timestamps, now)
return true, 0
}
retryAfter := rws.Timestamps[0] - windowStart
return false, retryAfter
}sourcecraft.dev
Сам алгоритм устроен несложно, мы просто отслеживаем, какие временные метки уже вышли за временное окно, и если оставшихся меток меньше, чем лимит запросов, клиент может выполнить очередной запрос. Иначе нужно ответить, через какое количество секунд освободится хотя бы одно место.
Далее подвязываем данную функцию к бэкенду:
func (store *InMemoryRollingWindowStore) CheckAndIncrement(
_ context.Context,
key string,
windowSize,
maxRequests,
now int64,
) (allowed bool, retryAfter int64, err error) {
err = store.Backend.Update(
key,
func() any { return &RollingWindowState{Timestamps: make([]int64, 0, 8)} },
func(v any) error {
st, ok := v.(*RollingWindowState)
if !ok {
return fmt.Errorf("unexpected state type %T for key %q", v, key)
}
allowed, retryAfter = st.check(windowSize, maxRequests, now)
return nil
})
return allowed, retryAfter, err
}sourcecraft.dev
То же самое в Redis:
var rollingWindowScript = redis.NewScript(`
local requestId = ARGV[1]
local windowSize = tonumber(ARGV[2])
local maxRequests = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
local windowStart = now - windowSize
redis.call("ZREMRANGEBYSCORE", KEYS[1], "-inf", windowStart)
local count = redis.call("ZCARD", KEYS[1])
if count < maxRequests then
redis.call("ZADD", KEYS[1], now, requestId)
redis.call("EXPIRE", KEYS[1], windowSize)
return {1, 0}
else
local oldest = redis.call("ZRANGE", KEYS[1], 0, 0, "WITHSCORES")
local retryAfter = 0
if oldest and oldest[2] then
retryAfter = tonumber(oldest[2]) - windowStart
end
return {0, retryAfter}
end
`)
func (s *RedisRollingWindowStore) CheckAndIncrement(ctx context.Context, key string, windowSize, maxRequests, now int64) (bool, int64, error) {
requestId := uuid.New().String()
return s.Backend.RunLuaScript(ctx, rollingWindowScript, []string{key}, requestId, windowSize, maxRequests, now)
}sourcecraft.dev
В Redis реализации стоит обратить внимание на то, что мы генерируемrequestId, всё потому, что мы храним наши элементы в множестве, а в множестве могут храниться только уникальные элементы. Но как тогда быть в ситуации, когда 2 и более запросов прилетают в одну секунду? Для этого и создан requestId, он добавляет уникальность каждому запросу, поэтому если в одну секунду прилетит 1000 запросов, они все посчитаются как разные запросы.
А что с метриками?
Как я ранее говорил, rate-limiter поддерживает подсчёт метрик. Начнём с функции instrument, о которой говорилось ранее:
package metrics
import (
"context"
"qwrttqr-rate-limiter/core/server/algos"
"time"
)
type instrumented struct {
next algos.Limiter
rec Recorder
}
func Instrument(next algos.Limiter, rec Recorder) algos.Limiter {
return &instrumented{next: next, rec: rec}
}
func (i *instrumented) Allow(ctx context.Context, req algos.Request) (algos.Decision, error) {
start := time.Now()
dec, err := i.next.Allow(ctx, req)
elapsed := time.Since(start)
switch {
case err != nil:
i.rec.ObserveDecision(Errored, elapsed)
case dec.Allowed:
i.rec.ObserveDecision(Allowed, elapsed)
default:
i.rec.ObserveDecision(Denied, elapsed)
}
return dec, err
}sourcecraft.dev
Данная функция просто возвращает структуру, которая обладает методом Allow, таким образом мы подменяем наш оригинальный limiter обёрткой, которая поддерживает подсчёт метрик.
Recorder реализован так:
package metrics
import "time"
type Result uint8
const (
Allowed Result = iota
Denied
Errored
NumResults
)
type Recorder interface {
ObserveDecision(r Result, elapsed time.Duration)
}sourcecraft.dev
И конкретная реализация данного интерфейса под Prometheus выглядит так:
package metrics
import (
"time"
"github.com/prometheus/client_golang/prometheus"
)
type PromRecorder struct {
decisions [NumResults]prometheus.Counter
durationHist [NumResults]prometheus.Observer
}
func NewPromRecorder(reg prometheus.Registerer) *PromRecorder {
counterVec := prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: "ratelimiter",
Name: "decisions_total",
Help: "Each decision count by its type.",
}, []string{"result"})
reqDurHistVec := prometheus.NewHistogramVec(prometheus.HistogramOpts{
Namespace: "ratelimiter",
Name: "decision_duration_seconds",
Help: "Execution time of one decision in seconds.",
Buckets: []float64{0.00005, 0.0001, 0.0005, 0.001, 0.1, 0.5, 1, 2},
}, []string{"result"})
reg.MustRegister(counterVec)
reg.MustRegister(reqDurHistVec)
p := &PromRecorder{}
p.decisions[Allowed] = counterVec.WithLabelValues("allowed")
p.decisions[Denied] = counterVec.WithLabelValues("denied")
p.decisions[Errored] = counterVec.WithLabelValues("errored")
p.durationHist[Allowed] = reqDurHistVec.WithLabelValues("allowed")
p.durationHist[Denied] = reqDurHistVec.WithLabelValues("denied")
p.durationHist[Errored] = reqDurHistVec.WithLabelValues("errored")
return p
}
func (p *PromRecorder) ObserveDecision(r Result, dur time.Duration) {
p.decisions[r].Inc()
p.durationHist[r].Observe(dur.Seconds())
}sourcecraft.dev
Здесь мы создаём счётчики и гистограммы под каждый результат rate-limiter'а и реализуем ObserveDecision, который их обновляет.
gRPC и REST
gRPC
Пишем .proto файл:
syntax = "proto3";
package ratelimiter.v1;
option go_package = "qwrttqr-rate-limiter/gen/ratelimiter/v1;ratelimiterv1";
import "google/protobuf/duration.proto";
service RateLimiter {
rpc Allow(AllowRequest) returns (AllowResponse);
}
message AllowRequest {
string key = 1;
double tokens = 2;
}
message AllowResponse {
bool allowed = 1;
google.protobuf.Duration retryAfter = 2;
}sourcecraft.dev
Вызываем protoc и остаётся просто реализовать полученные интерфейсы:
protoc -I proto --go_out=. --go_opt=module=qwrttqr-rate-limiter --go-grpc_out=. --go-grpc_opt=module=qwrttqr-rate-limiter ratelimiter/v1/ratelimiter.protosourcecraft.dev
package facade
import (
"context"
"qwrttqr-rate-limiter/core/server/algos"
ratelimiterv1 "qwrttqr-rate-limiter/gen/ratelimiter/v1"
"time"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/durationpb"
)
type GRPCServer struct {
ratelimiterv1.UnimplementedRateLimiterServer
l algos.Limiter
}
func NewGRPCServer(l algos.Limiter) *GRPCServer {
return &GRPCServer{l: l}
}
func (s *GRPCServer) Allow(ctx context.Context, in *ratelimiterv1.AllowRequest) (*ratelimiterv1.AllowResponse, error) {
tokens := in.GetTokens()
if tokens == 0.0 {
tokens = 1.0
}
req, err := NewRequest(in.GetKey(), tokens)
if err != nil {
return nil, status.Error(codes.InvalidArgument, err.Error())
}
d, err := s.l.Allow(ctx, req)
if err != nil {
return nil, status.Error(codes.Unavailable, "rate limiter error")
}
return &ratelimiterv1.AllowResponse{
Allowed: d.Allowed,
RetryAfter: durationpb.New(time.Duration(d.RetryAfter) * time.Second),
}, nil
}sourcecraft.dev
REST
Для REST всё совсем просто:
package facade
import (
"net/http"
"qwrttqr-rate-limiter/core/server/algos"
"strconv"
)
func HealthHandler(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
w.Write([]byte("ok"))
}
func LimitHandler(l algos.Limiter) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
key := r.Header.Get("X-Client-Key")
tokens := 1.0
if s := r.Header.Get("X-Required-Tokens"); s != "" {
n, err := strconv.ParseFloat(s, 64)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
tokens = n
}
req, err := NewRequest(key, tokens)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
d, err := l.Allow(r.Context(), req)
switch {
case err != nil:
http.Error(w, "rate limiter error", http.StatusInternalServerError) // store failure
case !d.Allowed:
w.Header().Set("Retry-After", strconv.FormatInt(d.RetryAfter, 10))
http.Error(w, "rate limit exceeded", http.StatusTooManyRequests)
}
}
}sourcecraft.dev
Middleware
Для реализации поддержки middleware можно использовать такую схему:
Приложение создаёт себе бэкенд, используя Redis или оставаясь на in-memory кэше
Используя созданный бэкенд создаём себе rate-limiter, с нужным алгоритмом
package middleware
import (
"qwrttqr-rate-limiter/core/config"
"qwrttqr-rate-limiter/core/server/algos"
"qwrttqr-rate-limiter/core/server/cache"
"time"
"github.com/redis/go-redis/v9"
)
var realClock = func() int64 { return time.Now().Unix() }
type Backend interface {
TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error)
FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error)
RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error)
}
type memoryBackend struct {
mem cache.InMemoryStoreInterface
}
func NewInMemoryBackend(cfg config.InMemoryBackendSettings) Backend {
return &memoryBackend{mem: cache.NewCache(
cfg.DefaultTtl,
time.Duration(cfg.EvictionTime)*time.Second,
cfg.ShardsCount),
}
}
type redisBackend struct {
redis cache.RedisStoreInterface
}
func NewRedisBackend(client *redis.Client, redisDefaultKeyTTL int64) Backend {
return &redisBackend{redis: cache.NewRedisStore(client, redisDefaultKeyTTL)}
}
func newTokenBucket(cfg config.AlgoSettings, store algos.TokenBucketStore) (algos.Limiter, error) {
if err := algos.ValidateTokenBucketConfig(cfg); err != nil {
return nil, err
}
return &algos.TokenBucketLimiter{
Capacity: *cfg.Capacity, Rate: *cfg.Rate, Store: store, Now: realClock,
}, nil
}
func newFixedWindow(cfg config.AlgoSettings, store algos.FixedWindowStore) (algos.Limiter, error) {
if err := algos.ValidateFixedWindowConfig(cfg); err != nil {
return nil, err
}
return &algos.FixedWindowLimiter{
WindowSize: *cfg.WindowSize, MaxRequests: *cfg.MaxRequests, Store: store, Now: realClock,
}, nil
}
func newRollingWindow(cfg config.AlgoSettings, store algos.RollingWindowStore) (algos.Limiter, error) {
if err := algos.ValidateRollingWindowConfig(cfg); err != nil {
return nil, err
}
return &algos.RollingWindowLimiter{
WindowSize: *cfg.WindowSize, MaxRequests: *cfg.MaxRequests, Store: store, Now: realClock,
}, nil
}
func (b *memoryBackend) TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error) {
return newTokenBucket(cfg, &algos.InMemoryTokenBucketStore{Backend: b.mem})
}
func (b *memoryBackend) FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error) {
return newFixedWindow(cfg, &algos.InMemoryFixedWindowStore{Backend: b.mem})
}
func (b *memoryBackend) RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error) {
return newRollingWindow(cfg, &algos.InMemoryRollingWindowStore{Backend: b.mem})
}
func (b *redisBackend) TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error) {
return newTokenBucket(cfg, &algos.RedisTokenBucketStore{Backend: b.redis})
}
func (b *redisBackend) FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error) {
return newFixedWindow(cfg, &algos.RedisFixedWindowStore{Store: b.redis})
}
func (b *redisBackend) RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error) {
return newRollingWindow(cfg, &algos.RedisRollingWindowStore{Backend: b.redis})
}sourcecraft.dev
Далее остаётся написать функцию, создающую непосредственно саму middleware, а именно функцию возвращающую тип, поддерживающий serveHTTP метод:
package middleware
import (
"log"
"net/http"
"qwrttqr-rate-limiter/core/server/algos"
"qwrttqr-rate-limiter/core/server/facade"
"strconv"
)
func NewHTTPMiddleware(
algo algos.Limiter,
keyDer HTTPKeyDerivationFunction,
costDer HTTPCostDerivationFunction,
failOpen bool,
) func(http.Handler) http.Handler {
if algo == nil || keyDer == nil {
panic("middleware: limiter and key function are required")
}
return func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
fail := func(stage string, err error) {
log.Printf("limiter error on %s: %v", stage, err)
if failOpen {
next.ServeHTTP(w, r)
return
}
http.Error(w, "service unavailable", http.StatusServiceUnavailable)
}
tokens := 1.0
if costDer != nil {
t, err := costDer(r)
if err != nil {
fail("cost derivation", err)
return
}
tokens = t
}
key, err := keyDer(r)
if err != nil {
fail("key derivation", err)
return
}
req, err := facade.NewRequest(key, tokens)
if err != nil {
fail("request creation", err)
return
}
decision, err := algo.Allow(r.Context(), req)
if err != nil {
fail("allow", err)
return
}
if !decision.Allowed {
w.Header().Set("Retry-After", strconv.FormatInt(decision.RetryAfter, 10))
http.Error(w, "rate limit exceeded", http.StatusTooManyRequests)
return
}
next.ServeHTTP(w, r)
})
}
}sourcecraft.dev
С gRPC необходимо реализовать 2 вида interceptor'а - унарный и потоковый:
func NewGRPCUnaryInterceptor(
algo algos.Limiter,
keyDer GRPCKeyDerivationFunction,
costDer GRPCCostDerivationFunction,
failOpen bool,
) grpc.UnaryServerInterceptor {
check := newGRPCCheck(algo, keyDer, costDer, failOpen)
return func(
ctx context.Context,
req any,
info *grpc.UnaryServerInfo,
handler grpc.UnaryHandler,
) (any, error) {
if err := check(ctx); err != nil {
return nil, err
}
return handler(ctx, req)
}
}
func NewGRPCStreamInterceptor(
algo algos.Limiter,
keyDer GRPCKeyDerivationFunction,
costDer GRPCCostDerivationFunction,
failOpen bool,
) grpc.StreamServerInterceptor {
check := newGRPCCheck(algo, keyDer, costDer, failOpen)
return func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
if err := check(ss.Context()); err != nil {
return err
}
return handler(srv, ss)
}
}sourcecraft.dev
И check:
func newGRPCCheck(
algo algos.Limiter,
keyDer GRPCKeyDerivationFunction,
costDer GRPCCostDerivationFunction,
failOpen bool,
) func(ctx context.Context) error {
if algo == nil || keyDer == nil {
panic("middleware: limiter and key function are required")
}
return func(ctx context.Context) error {
fail := func(stage string, err error) error {
log.Printf("limiter error on %s: %v", stage, err)
if failOpen {
return nil
}
return status.Error(codes.Unavailable, "service unavailable")
}
tokens := 1.0
if costDer != nil {
t, err := costDer(ctx)
if err != nil {
return fail("cost derivation", err)
}
tokens = t
}
key, err := keyDer(ctx)
if err != nil {
return fail("key derivation", err)
}
req, err := facade.NewRequest(key, tokens)
if err != nil {
return fail("request creation", err)
}
decision, err := algo.Allow(ctx, req)
if err != nil {
return fail("allow", err)
}
if !decision.Allowed {
md := metadata.Pairs("retry-after", strconv.FormatInt(decision.RetryAfter, 10))
_ = grpc.SetTrailer(ctx, md)
return status.Error(codes.ResourceExhausted, "rate limit exceeded")
}
return nil
}
}sourcecraft.dev
Как вы могли заметить, везде можно передать функции извлечение ключа и стоимости запроса. Это нужно так как в режиме middleware вы чаще всего будете писать сервисы, которые смотрят наружу, а значит будут принимать запросы от недоверенных клиентов.
Итог
Статья получилась достаточно объёмной, надеюсь она будет полезна ребятам, которые только начинают учить Golang)
исходник
Two Data Sources, Zero Orchestrators — Building an F1 Analytics Pipeline on AWS Glue
> Disclaimer: I'm not affiliated with or endorsed by AWS. This is a personal hobby project, and some of the behaviour described here was found by trial and error rather than from the docs. Spark Declarative Pipelines is new in AWS Glue 6.0 and may change, so check the official AWS Glue documentation first.
In my first Spark Declarative Pipelines post I walked through the basic idea: declare the tables you want, let AWS Glue 6.0 work out the dependency order, and skip the orchestrator entirely. That demo used one CSV source flowing bronze → silver → gold. It worked, but it was a tidy example — one source, one shape, one straight line.
Real pipelines are messier. So I built one properly, using Formula 1 race data, and deliberately made it harder: data arriving from a live API as deeply nested JSON, a second completely unrelated data source, and dimension tables that need to be deduplicated. This post walks through that build, and the things I only learned by breaking it.
New to SDP? The short version: Spark Declarative Pipelines (SDP) is a framework in AWS Glue 6.0 where you declare tables instead of writing scripts that build them in order. You reference one table from another by name, and the engine figures out the dependency graph, the execution order, and what can run in parallel — all inside a single Glue job. My first post covers the fundamentals if you want them.
First, the use case: what are we actually building?
Let's park AWS for a minute and talk about the data, because the data shape is what drives every design decision later.
I'm using f1api.dev, a free public Formula 1 API. You ask it for one race at a time:
GET https://f1api.dev/api/2024/1/race
That's season 2024, round 1 — the Bahrain Grand Prix. What comes back is a single, deeply nested JSON document describing everything about that race. Trimmed down heavily for readability:
{
"season": 2024,
"races": {
"raceName": "Bahrain Grand Prix",
"date": "2024-03-02",
"round": "1",
"circuit": [
{
"circuitId": "bahrain",
"circuitName": "Bahrain International Circuit",
"city": "Sakhir",
"country": "Bahrain",
"corners": 15
}
],
"results": [
{
"position": "1",
"grid": "1",
"points": 26,
"driver": { "driverId": "max_verstappen", "name": "Max", "surname": "Verstappen", "number": 1 },
"team": { "teamId": "red_bull", "teamName": "Red Bull Racing" }
},
{ "position": "2", "points": 18, "driver": { "driverId": "perez", "...": "..." }, "...": "..." }
]
}
}
Look at the shape, because this is the first real problem:
- Race facts live at the top (name, date, round)
- The circuit is an array nested one level down
- The results are an array of 20 entries — one per driver — and inside each entry there's a nested driver object and a nested team object
You cannot answer "which driver scored the most points this season?" from a folder full of documents shaped like that. You need flat tables. That flattening is the pipeline's first job.
The questions I want answered
The whole point of the pipeline is to produce tables that answer questions like:
- Which driver scored the most points, won the most races, and had the best average finishing position?
- Which team scored the most, and how many different drivers did they field?
- Which circuits have been raced at, how many times, and what's the lap record there?
Why three layers?
This is the medallion architecture — bronze, curated (often called silver), and gold. It sounds formal, but the idea is simple: don't try to go from raw JSON to a polished dashboard table in one step. Do it in three, where each step has one job.
Bronze — land it, then flatten it.
First, raw_race stores the API response more or less as it arrived. One row per pipeline run. Nothing cleaned, nothing dropped. If I later discover a bug in my parsing logic, the original data is still sitting here and I can reprocess it.
Then four tables flatten that nesting into something usable:
| Table | What one row means |
|---|---|
| race_results_unnested | One driver's result in one race (20 rows per race) |
| race_drivers | A driver's profile, as it appeared in that race |
| race_teams | A team's profile, once for each of its drivers in that race |
| race_circuits | The circuit that race was held at |
!Bronze layer ER diagram: raw_race flattened into four prepared tables, with circuit_track_type shown unconnected
Curated — one row per thing.
Here's the catch with that bronze layer: every single race response re-sends the full profile of every driver and team. Load 24 races and you have 24 copies of Max Verstappen's profile and about 48 copies of Red Bull's team record (one for each of its two drivers, every race) — most of them identical, some slightly different as the season progresses and championship counts change.
For analytics I want exactly one current row per driver. So the curated layer deduplicates:
- dim_drivers — one row per driver (the most recent version of their profile)
- dim_teams — one row per team
- dim_circuits — one row per circuit
- fact_race_results — one row per (race, driver): the actual measurements — grid position, finishing position, points scored
If you've seen a star schema before, that's exactly what this is: dimensions describe the who/what/where, and the fact table records what happened, with the numbers.
!Curated layer ER diagram: three dimension tables joined to the fact_race_results table
Gold — the answers.
Three summary tables, each one a direct answer to one of the questions above: driver_performance_summary, team_performance_summary, and circuit_summary. These are what a dashboard or an Athena query would actually read.
!Gold layer ER diagram: three standalone summary tables
One run = one race
One design decision worth calling out because it shapes everything: each pipeline run fetches exactly one race. You tell it the season and round when you start the job:
scripts/run_pipeline.sh --profile my-profile --season 2024 --round 1 --mode run --wait
Run it again with --round 2 and the pipeline accumulates — bronze grows by one more race, and the curated and gold layers recompute to include it. After running rounds 1 and 2, my gold table showed Verstappen on 2 wins and 51 points, Perez on 36, Leclerc on 28 — which matches the real 2024 championship standings after two races — always a nice moment when the numbers come out right.
!Athena query results: driver standings after two races, Verstappen first on 51 points
The pipeline at a glance
All of the above runs as one Glue job. Not one job per layer, not a Step Functions state machine tying them together — one job, with all tables registered as Apache Iceberg tables in the Glue Data Catalog.
!Glue console Runs tab for f1-sdp-job showing a successful run and its input arguments.The red box shows the job's arguments: --enable-spark-declarative-pipeline is what switches SDP on, and the --conf values are the season and round I passed for this run.
In SDP you write two kinds of tables, and the difference matters:
- Streaming tables process only what's new since the last run. Bronze uses these, so re-running never reprocesses an old race.
- Materialized views fully recompute every run. Curated and gold use these, because "the latest row per driver" and "total points per driver" genuinely need to look at everything each time.
!Glue Data Catalog table list for f1_sdp_db showing 13 tables
Adding a second data source (and why F1 needed one)
Here's where I actually tested the "zero orchestration" claim.
The F1 API tells me a circuit's name, city, country, length, and corner count. It does not tell me whether it's a street circuit (Monaco, Singapore, Las Vegas — public roads closed for the weekend) or a permanent racetrack (Silverstone, Spa, Suzuka — purpose-built facilities). That's exactly the kind of attribute you'd want to slice results by — for example, comparing how drivers perform on street circuits versus permanent ones.
That data simply isn't in the API. It has to come from somewhere else — which makes it a perfect second data source, and a genuinely different kind of source:
| | raw_race | circuit_track_type |
|---|---|---|
| Where it comes from | Live external API | Static CSV files in S3 |
| How it arrives | An HTTP call during the job run | Uploaded by a deploy script |
| Depends on | Nothing — it's a root | Nothing — also a root |
Two completely independent branches. They never touch in bronze. They meet for the first time in the curated layer, in exactly one place:
CREATE MATERIALIZED VIEW dim_circuits AS
SELECT rc.circuit_Id, rc.name, rc.city, rc.country, ctt.track_type, rc.length, rc.corners, ...
FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY circuit_Id ORDER BY race_Date DESC, load_date_time DESC) AS _rn
FROM race_circuits
) rc
LEFT JOIN (
SELECT *, ROW_NUMBER() OVER (PARTITION BY circuit_Id ORDER BY load_date_time DESC) AS _rn
FROM circuit_track_type
) ctt ON rc.circuit_Id = ctt.circuit_Id AND ctt._rn = 1
WHERE rc._rn = 1;
Ignore the ROW_NUMBER() noise for a second (that's the deduplication I mentioned — "keep the newest row per circuit"). The important line is LEFT JOIN ... circuit_track_type.
That join is the entire integration. I didn't register the new source anywhere. No config file entry, no trigger, no second job, no DAG definition. SDP reads this SQL file, sees a reference to a table called circuit_track_type, and from that alone works out that it's an upstream dependency of dim_circuits. The two branches stay independent until this join — and per the AWS docs, SDP runs independent branches in parallel.
That's the claim from my first article, and this is the first time I'd properly stress-tested it. It held.
!Athena query results: bahrain is a permanent circuit and jeddah a street circuit
Four lessons the tidy demo didn't teach me
A one-source demo hides most of the real friction. These four cost me the most time. Here they are at a glance, then in detail.
| # | The lesson | What I saw | What worked |
|---|---|---|---|
| 1 | No MERGE or upsert | No built-in way to keep a row's history | Keep only the latest row per key (SCD Type 1) |
| 2 | Streaming tables need a streaming source | An INVALID_FLOW_QUERY_TYPE error on the API data | Stage the API response in S3, then read it as a stream |
| 3 | Fresh deployments can fail | PATH_NOT_FOUND before any data had landed | Write a tiny .keep marker file into each watched folder |
| 4 | Shared code isn't importable | ModuleNotFoundError | One shared bootstrap file, listed first in the manifest |
1. There's no MERGE or upsert, so there's no change history
The problem. Databricks' version of declarative pipelines has a built-in feature for tracking how dimension rows change over time (create_auto_cdc_flow). AWS Glue 6.0 SDP has no equivalent.
Why. SDP gives you only two kinds of persisted table to write to: append-only streaming tables and full-recompute materialized views. There's no "update this row" operation.
What worked. I settled for SCD Type 1: keep only the current version of each row, with no history. (SCD Type 2 would keep every past version.) Every curated table works out "latest row per key" with a ROW_NUMBER() window function over the whole bronze history — number each driver's rows newest-first, then keep row 1. It works well. But if you genuinely need history ("what was this team's championship count last month?"), know that going in: there's no built-in escape hatch yet.
2. A streaming table can't be fed from a plain API call
The problem. My first instinct was the obvious one: fetch the JSON from the API, load it into a DataFrame in memory, and return it from the pipeline. Glue rejected it:
AnalysisException: [INVALID_FLOW_QUERY_TYPE.BATCH_RELATION_FOR_STREAMING_TABLE]
Why. A streaming table has to be fed by a streaming source: something Spark can read incrementally while remembering how far it got. A DataFrame built in memory is a one-off batch, so it's refused.
What worked. Stage the data in S3 first:
1. Call the API and get the JSON.
2. Write it to a uniquely named file in an S3 folder, using plain boto3 (not Spark).
3. Read that folder with spark.readStream.
Now it's a genuine streaming source, and Spark's built-in file checkpointing means a re-run never reprocesses yesterday's file. It's an extra moving part that exists purely because a REST API has no natural streaming shape.
3. A brand-new deployment fails before it does anything
The problem. On a fresh deploy, the pipeline failed straight away:
AnalysisException: [PATH_NOT_FOUND] Path does not exist: s3://.../reference-data/circuit_track_type/
Why. Both S3 folders the pipeline watches (the API staging folder and the reference-data folder) are read with spark.readStream(...).load(path). Before any data has landed, those folders don't exist, and Spark doesn't shrug and read zero rows. It fails the job.
What worked. Make sure each folder always exists by writing a tiny empty marker file, .keep, into it. My CDK stack does this for the reference-data folder on every deploy, and the pipeline does it itself for the API staging folder. It's cheap insurance against a failure that only appears on day one of a new environment, which is exactly when you're least expecting it.
4. Shared Python code isn't importable by default
The problem. I keep schema definitions and API-fetch helpers in a shared Python package next to the transformation files. Importing it failed:
ModuleNotFoundError: No module named 'f1_pipeline_lib'
Why. SDP loads each transformation file directly by its file path, and that doesn't add anything to Python's import path (I confirmed this by reading SDP's source). I then tried the documented way to ship extra Python files to a Glue job, the --extra-py-files parameter. It failed with the same error, most likely because SDP runs as a separate process that the setting doesn't reach. I confirmed both failures by running real Glue jobs and reading the tracebacks.
What worked. One tiny shared bootstrap file that fixes the import path, listed first in the pipeline manifest so it always runs before anything else. One file, one place, instead of the same few lines copy-pasted into every transformation file.
Benefits of using SDP in AWS Glue 6.0
- The zero-orchestration claim genuinely holds, even with two independent sources and a deduplicating curated layer. One job, no Step Functions, no Airflow.
- Connecting a new source takes one line. The source still needs its own table definition, but wiring it into the pipeline is just a SQL JOIN or a .readStream.table("x") call — there's nothing to register anywhere else.
- Independent branches run in parallel automatically, according to the AWS docs — no fan-out/fan-in code to write.
- SQL and Python mix freely in the same pipeline — my bronze layer is Python, curated and gold are plain SQL files.
- Streaming tables remember what they've already processed. Each one keeps a checkpoint in S3 — a record of which input it has already consumed — so every run picks up only what's new. I checked this directly on the reference-data table: after adding a second CSV file, the checkpoint's log showed the next run read just that new file, not the original again. The same mechanism means running round 2 adds that race alongside round 1 instead of starting over. It isn't free in the cost sense — you still pay for the compute that runs — but you don't write or maintain any "last processed" bookmark yourself, which is easy to get subtly wrong in a hand-rolled pipeline. (Materialized views don't get this; they recompute from scratch every run, which is the trade-off listed under limitations below.)
- Tables register themselves in the Glue Data Catalog — no manual CREATE TABLE anywhere.
- VALIDATE mode is a genuinely useful dry run that catches dependency and compile errors before you spend money on a full run.
Limitations of using SDP in AWS Glue 6.0
Three of these — materialized views always recomputing, no data quality expectations, and the .withColumn() gotcha — are also listed under Considerations and limitations in the official AWS docs.
- No CDC/MERGE/upsert. No SCD Type 2 out of the box, as described above.
- No data quality expectations. Inline data quality annotations are not supported in the current SDP framework. Some other declarative frameworks let you attach rules like "this column must not be null" directly to a table definition; SDP in AWS Glue 6.0 has no equivalent yet, so any validation is hand-written. My pipeline doesn't have a quality layer yet, so a malformed API response simply fails the run.
- Materialized views always fully recompute. Only streaming tables are incremental — worth budgeting for as your gold aggregations grow.
- Materialized views can read as empty while a job is running. Every run clears each materialized view first (a TRUNCATE TABLE), and the new rows only land when that view's own step finishes. So an Athena query during a run returns zero rows — not an error, and not the previous data — until the refresh completes. I checked this on a real run of this pipeline by polling Athena every few seconds: each of my four materialized views read 0 rows for roughly 30 seconds of a two-minute run, with the gold tables coming back last, while a streaming table stayed readable the whole time. I couldn't find this in the AWS Glue or Apache Spark documentation; I found the cause in Spark's open-source SDP code. In Spark 4.1.1, the version Glue 6.0 runs, it truncates every non-streaming table at the start of an update. Plan for it: don't point a dashboard or a downstream job at these tables while a run is in progress.
- External API sources need a staging workaround to satisfy the streaming-relation requirement.
- Python packaging has sharp edges that I couldn't find documented. The import-path issue, and the fact that --extra-py-files doesn't reach the SDP process, both cost me real debugging time.
- A .withColumn() gotcha. If a downstream table reads another pipeline table and adds a derived column with .withColumn(...), SDP might fail to detect the dependency on the second and later runs, so the downstream table reads stale data from the previous run (a one-run lag). The documented fix is to build derived columns inside a single .select(...) instead, and to avoid operations that force plan resolution, such as .schema or .collect, inside query functions. I followed this from the start — every derived column in my bronze layer is written in a .select(...) — so I never saw the stale-data problem itself. I did hit a similar sharp edge in the same place: adding .distinct() to a streaming read made the pipeline fail with a misleading "table not found" error, and removing it fixed it.
- VALIDATE still runs your code; it just doesn't write the results. A live-API source will happily call the real API during a "dry run" unless you guard against it.
What I took away from this
My first article proved the concept worked. This one was about whether it survives something less convenient — a second unrelated source, nested data that needs real flattening, and state that has to stay correct across many runs. It does, and the "add a source, write one join" story held up exactly as advertised. That still impresses me.
What surprised me is where the effort actually went. Almost none of it was in the declarative pipeline logic — that part really is as clean as promised. It all went into the handful of places where SDP's abstractions meet the realities of Glue and Spark underneath: the streaming-source rule, the missing-folder failure, and the import path. And none of those were things I could have read my way to. Every one came from running a real job, reading the actual CloudWatch traceback, and occasionally reading the framework's own source code to find out what it was really doing.
If you're evaluating SDP for anything beyond a demo, that's my main advice: budget time for that kind of hands-on verification. The gap between "the docs say" and "a real Glue 6.0 job does" is still a real gap today. Just know that the core promise — one job, no orchestrator, dependencies figured out for you — does deliver.
The code for this project is on GitHub.
References
- "One Glue Job, Zero Orchestrators — My First Spark Declarative Pipeline on AWS Glue", DEV Community
- "Build declarative ETL pipelines with AWS Glue 6.0", AWS Big Data Blog
- AWS Glue Spark Declarative Pipelines documentation
- f1api.dev — the free Formula 1 API used in this project
---
About the author
I'm an AWS Community Builder in the Data category. I build small, real projects and write up what the documentation didn't cover.
Questions, or something broke differently for you? Connect with me on LinkedIn or drop a comment below.
1 reaction
AUTO_CDC: Turning an event log into current state with Spark Declarative Pipelines
The question an event log can't answer cheaply
The dataset I use for most of my pipeline demos is a simulated ghost kitchen (a delivery-only restaurant operation), and like most operational data it arrives as an append-only log. Every order produces a lifecycle of events, each stamped with a time and a per-order sequence number, and a day of it is about a million events. The generator deliberately injects a few percent of late and duplicated events, because real streams do that.
Logs like this are excellent at answering what happened and awkward at answering what operations people actually ask: which orders are live right now, and where is each one?
The usual answer is to recompute it. Group the whole history by order, take the latest event for each, drop the cancelled ones, and repeat on every refresh, paying for the entire history every time and getting it subtly wrong whenever "latest" means latest to arrive rather than latest to happen.
The better answer depends on noticing something that sounds obvious once it's said: an append-only lifecycle log is already a change data capture feed. Each event is a change to the current state of one entity (the order), keyed by order id, sequenced by when it happened, and a cancellation is simply a delete.
Here is an illustrative order (simplified; the real sequence is described below), first as the lifecycle log records it, then as the change feed it implies:
| seq | event_type | as a change to orders_live |
|---|---|---|
| 1 | order_created | UPSERT order 1042, status = order_created |
| 2 | kitchen_started | UPSERT order 1042, status = kitchen_started |
| 3 | order_ready | UPSERT order 1042, status = order_ready |
| 4 | driver_picked_up | UPSERT order 1042, status = driver_picked_up |
| 5 | driver_ping | UPSERT order 1042, status = driver_ping |
| 6 | delivered | UPSERT order 1042, status = delivered |
An order that ends in order_cancelled produces a DELETE as its last change instead, and disappears from the table. Once the log is seen this way, "current status of every order" stops being an aggregation problem and becomes the problem of applying a change feed to a table, which is exactly what AUTO CDC was built to do.
My argument is that AUTO CDC is worth adopting less because it saves code (though it does, a lot) than because it moves the hard parts of change application into the engine, and leaves you with a small, explicit contract: a sequence column that is unique and never null, a target that genuinely supports row-level operations, and deletes expressed as events. Most of what goes wrong with hand-rolled CDC is the engine's side of that contract done badly; most of what can still go wrong with AUTO CDC is your side of it.
What AUTO CDC is, and where it stands
AUTO CDC began as APPLY CHANGES in Databricks' Delta Live Tables and was contributed to Apache Spark under its own SPIP (SPARK-56249), as part of Spark Declarative Pipelines. In Python it's create_auto_cdc_flow.
| Release | AUTO CDC status |
|---|---|
| Spark 4.2.0 (July 2026) | Shipped. SCD Type 1 via the Python / Spark Connect API |
| Spark 4.3.0 (release candidate) | Adds SCD Type 2 (history with validity ranges) |
| main branch (5.0.0-SNAPSHOT) | Ongoing work under SPARK-56249 |
Everything I tested was SCD Type 1 (the target holds only the current state per key, and updates overwrite in place), on a build of the main branch from late May 2026, before 4.2 was released.
Applying a change feed correctly means doing several things at once:
- matching each change to its row by key
- keeping the version with the highest sequence per key, not the last to arrive
- honouring deletes, but only when the delete is the latest thing that happened to that key
- staying idempotent when a streaming source redelivers after a failure
The hand-rolled version is usually a foreachBatch with windowed deduplication feeding a MERGE, and it tends to get the cross-batch cases wrong without throwing errors. A late update that lands in the micro-batch after a newer one quietly overwrites it, and nobody notices until the numbers disagree with something else.
!diagram 1
The code
The projection from lifecycle log to change feed is ordinary Spark. This is the core of what my test harness does: every event becomes an UPSERT of the order's current status, a cancellation becomes a DELETE, and the sequence is made unique per order.
feed = events.select(
F.col("order_id"),
F.col("event_type").alias("status"),
F.col("location_id").cast("int").alias("location_id"),
F.col("ts_seconds").alias("event_ts"),
F.when(F.col("event_type") == "order_cancelled", F.lit("DELETE"))
.otherwise(F.lit("UPSERT")).alias("op"),
# unique per-key ordering: event time, broken by per-order sequence
(F.col("ts_seconds") * F.lit(100) + F.col("sequence")).alias("seq"),
)
Then the pipeline itself is three declarations: a streaming source, a target streaming table, and the flow.
from pyspark import pipelines as dp
@dp.table(name="order_cdc_source")
def order_cdc_source():
return spark.readStream.schema(ORDER_SCHEMA).json("file:///data/order_cdc/*.json")
dp.create_streaming_table("ice.gold.orders_live")
dp.create_auto_cdc_flow(
target="ice.gold.orders_live",
source="order_cdc_source",
keys=["order_id"], # how a change finds its row
sequence_by="seq", # highest sequence per key wins, whatever the arrival order
apply_as_deletes="op = 'DELETE'", # which events delete the row
except_column_list=["op", "seq"], # bookkeeping columns to leave out of the target
stored_as_scd_type=1,
)
The engine handles what the hand-rolled version gets wrong: deduplication across micro-batches by sequence_by, the generated upsert and delete, exactly-once checkpointing, and the auxiliary state table that remembers the latest sequence per key across runs.
That table is what lets a late event arriving in a later batch resolve correctly.
| | Recompute from the log | AUTO CDC |
|---|---|---|
| Cost per refresh | re-aggregates the entire history | processes only new events |
| Out-of-order events | "latest" by arrival is wrong | resolved by sequence_by |
| Cancellations | a row every query must remember to filter | the row is deleted |
| Code | bespoke foreachBatch + dedup + MERGE | three declarations |
Running it on messy data
I ran the order-status projection on a sample of the ghost-kitchen stream: a little under five thousand orders and roughly 125,000 of their lifecycle events, including the injected late and duplicate ones. AUTO CDC collapsed them into one row per live order:
| gold.orders_live | rows |
|---|---|
| delivered | 4,601 |
| driver_ping (mid-delivery) | 84 |
| order_ready | 3 |
| order_created | 1 |
| kitchen_started | 1 |
| driver_picked_up | 1 |
| total live orders | 4,691 |
| cancelled orders removed by apply_as_deletes | 308 |
To check it, I recomputed the same table from scratch the naive way and compared the two in both directions:
w = Window.partitionBy("order_id").orderBy(F.desc("seq"))
naive = (feed.withColumn("rn", F.row_number().over(w)).filter("rn = 1")
.filter("op != 'DELETE'").select("order_id", "status"))
pairs = spark.table("ice.gold.orders_live").select("order_id", "status")
match = naive.subtract(pairs).count() == 0 and pairs.subtract(naive).count() == 0
# match = True
They matched exactly.
It's worth being precise about what that proves: it compares the two columns the table exists to answer, as sets, rather than every column byte for byte.
Two details of the run also deserve honesty, because they're the kind of thing that makes a demo look cleaner than it was. The sample was drawn by sorting order ids and taking the first five thousand, and because the raw data contains events with a null order id (which Spark sorts first), the sample was really 4,999 orders.
And the projection joined each event to its order's creation event, so orders with duplicated creation events fanned out into duplicate rows in the feed. AUTO CDC handled both without complaint, since duplicate copies of an event carry the same sequence value and resolve to the same row, but a pipeline that quietly processes inputs it wasn't meant to receive is a reminder to look at your inputs before trusting your outputs.
The second scenario was the textbook one: keeping a dimension current as kitchens relocate, rename, and close. These are the exact changes streamed into silver.dim_locations_current, on top of an initial load of four locations at sequence 1:
changes = [
{"id": 2, "city": "Los Angeles - Arts District", "op": "UPSERT", "seq": 5},
{"id": 1, "city": "San Francisco - SoMa", "op": "UPSERT", "seq": 4},
{"id": 3, "city": "Chicago", "op": "DELETE", "seq": 6}, # kitchen closed
{"id": 2, "city": "Los Angeles (STALE)", "op": "UPSERT", "seq": 3}, # arrives late, older
]
| id | after AUTO CDC | why |
|---|---|---|
| 1 | San Francisco - SoMa | rename at seq 4 applied |
| 2 | Los Angeles - Arts District | rename at seq 5 kept; the stale seq 3 edit ignored |
| 3 | (no row) | closure at seq 6 deleted it |
| 4 | Austin | unchanged |
A nightly full reload would eventually reach the same names, but it copies every unchanged row to get there and has no way at all to express that a location no longer exists.
That's the more general lesson: an append-only or aggregate pipeline can only add facts, so "this was cancelled" becomes another row every downstream query has to remember to filter. In a CDC pipeline a terminal business event can be the delete itself, and the same mechanism propagates hard deletes for compliance.
Edge cases, and how Type 1 resolves them
I put SCD Type 1 through a battery of adversarial cases, and it behaved correctly in every one. The behaviours are worth knowing precisely, because this is where intuition tends to be wrong. These are the cases from my stress harness, with the events in arrival order; | separates micro-batches.
| Case | Events for one key | Result |
|---|---|---|
| Upsert to a key that never existed | UPSERT@1 | row inserted |
| Delete of a key that never existed | DELETE@1 | no-op |
| Several updates in one batch | UPSERT@1, UPSERT@2, UPSERT@3 | row has the @3 values |
| Stale delete | UPSERT@1 | UPSERT@3 | DELETE@2 | row stays, with the @3 values; the update is newer |
| Stale insert after a delete | UPSERT@3 | DELETE@5 | UPSERT@2 | row stays deleted |
| Delete arrives before an older insert | DELETE@5 | UPSERT@2 | row stays deleted |
| Resurrection | UPSERT@1 | DELETE@2 | UPSERT@3 | row present, with the @3 values |
| Volume, out of order | 500 keys, each UPSERT@1, @3, @2 shuffled across 5 batches, plus DELETE@5 for 100 of them | exactly 400 rows, all with the @3 values |
| Tie (probe, not pass/fail) | UPSERT@5 "Boston", UPSERT@5 "Denver" in one batch | one of them, chosen arbitrarily |
Your side of the contract
The sequence. A null in the sequence_by column fails the micro-batch with AUTOCDC_MICROBATCH_VALIDATION.NULL_SEQUENCE rather than guessing, which is the right behaviour, and means the column has to be non-null by construction.
Ties are worse, because they don't fail: two changes to the same key with the same sequence in one micro-batch are resolved arbitrarily, as the programming guide itself warns. A coarse timestamp is the classic way to produce ties, so sequence on something strictly increasing per key, like a log offset or commit LSN. My composite of timestamp times 100 plus the per-order sequence works only because no order in this data has more than 99 events; it's a local trick, not a general recipe. The ties that did occur in my data were all injected exact duplicates, the one case where an arbitrary choice is harmless.
The target. AUTO CDC needs a connector that implements Spark's DataSource V2 row-level operation interface (SupportsRowLevelOperations), and "supports MERGE" is not the same thing:
| Target | Result in my testing |
|---|---|
| Plain Parquet | rejected immediately: AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE |
| Delta | did not work: Delta's table implementation doesn't expose SupportsRowLevelOperations, despite supporting MERGE elsewhere |
| Iceberg, Hadoop catalog | worked |
| Iceberg, JDBC (Postgres) catalog | worked |
| Iceberg, Hive Metastore catalog | worked |
| Iceberg, Gravitino REST catalog | worked |
| Iceberg, Glue catalog (emulated) | worked |
| Iceberg, Unity Catalog OSS REST | failed: the catalog is read-only |
| Any ephemeral / in-memory catalog | unsuitable: the aux state table must persist across restarts |
Two smaller surprises: the target carries a hidden __spark_autocdc_metadata column that shows up in SELECT *, and Iceberg's own changelog is batch-only in open-source Spark for now, so you can't stream one Iceberg table's changes straight into AUTO CDC as a source.
A correction. I originally reported that running an AUTO CDC pipeline through the spark-pipelines CLI failed at registration with InvalidPlanInput: RELTYPE_NOT_SET, and called it a bug. It wasn't.
The CLI connects to the default Spark Connect endpoint, and an old Spark 4.1 Connect server was still listening there, so my pipeline was being sent to a server that had never heard of AUTO CDC. Against a server of the right version the CLI works correctly. If you see that error, check what's actually listening on sc://localhost:15002 before anything else.
What changes
With a hand-rolled pipeline, the question you're left with is whether you've correctly implemented cross-batch deduplication, sequenced deletes, and idempotent merges, and the honest answer is usually that you won't know until something disagrees. With AUTO CDC that part belongs to the engine, which is tested against exactly the cases people get wrong.
The question becomes whether you've given it a sequence that is unique and never null, a target that really supports row-level operations, and a delete for everything that should be allowed to disappear. That's a much better question to be left with, and the first step is often not a new data source at all; it's noticing that the event log you already have has been a change feed all along.
Further reading
- Test harnesses and results: lakehouse-stack/scripts/pipelines/autocdc-scd1 (Spark main branch, 5.0.0-SNAPSHOT at 06ffa273, with an Iceberg runtime ported to Spark 5.0).
1 reaction · 3 comments
From Pickle to Arrow: A Decade of Moving Data Between Spark's JVM and Python
Arrow's real contribution to PySpark isn't speed at any one API but a shared memory format at every boundary between the JVM and Python; each release removed one conversion at one boundary, what each one left behind explains the next, and the conversion that remains now lives inside Spark's own engine rather than at the Python edge.
That crossing is the oldest problem in PySpark, and the most persistent attempt to fix it started in February 2016, when Wes McKinney (the creator of pandas) filed a Spark ticket with a modest title: SPARK-13534, "Implement Apache Arrow serializer for Spark DataFrame for use in DataFrame.toPandas." It asked for one method to stop using pickle.
It took two years to ship. And over the decade that followed, the same idea went through every other door between Spark's JVM and Python: UDFs, partition-level APIs, the network protocol of Spark Connect, custom data sources, until in Spark 4.1 a Python function could receive Spark's data as an Arrow array and never touch pandas at all, and in Spark 4.2 the Arrow paths finally became the defaults.
---
The wall
PySpark is not a Python reimplementation of Spark. It's a Python client for a JVM engine. When you write df.filter(col("x") > 5).groupBy("y").count(), Python builds a query plan and hands it to the JVM, which optimizes, distributes, and executes it; Python never touches the data, which is why DataFrame code runs at the same speed in PySpark and Scala.
The trouble starts the moment your own code needs the data: a UDF, toPandas(), createDataFrame(pandas_df), a custom data source. Then data has to physically cross from the JVM into a Python process, and for most of Spark's history the only way across was pickle, one row at a time:
!diagram 1
Every row pays four serialization steps and a function call. For a UDF like return x * 1.1, the arithmetic is nanoseconds and the machinery around it is microseconds, three orders of magnitude of overhead wrapped around almost no work.
That was PySpark from 2015 to 2018, and the same crossing existed at every door between the two runtimes, on both the driver and the executors:
!diagram 2
Why the fix was a memory format, not a faster pickle
Arrow was conceived in late 2015 by people who had hit the same wall from different directions. McKinney had spent years fighting pandas' memory model; his 2017 essay "Apache Arrow and the 10 Things I Hate About pandas" lists the grievances, including a rule of thumb that pandas users needed "5 to 10 times as much RAM as the size of your dataset."
Jacques Nadeau, building a query engine at Dremio, saw every data system inventing its own internal format and paying to translate between them at every boundary. Marcel Kornacker, who created Impala at Cloudera, saw the same copying across the Hadoop ecosystem. They met at Strata NYC in September 2015, and Arrow was announced as an Apache top-level project in February 2016, skipping the incubator entirely, which is unusual and reflected how many companies wanted the same thing.
The idea was not a faster serializer. It was to agree on the bytes themselves: a language-independent, columnar, in-memory layout specified down to byte alignment, null bitmaps, and buffer order, so that two Arrow-aware systems can hand each other data without converting it.
Within a process, or with shared memory, that's genuinely zero-copy; across a socket it's still a copy, but with no per-row decoding on either side, because the bytes on the wire are already the bytes in memory.
Columnar mattered as much as standard. In a row layout, values from one column are interleaved with everything else in the row:
row layout: [id=1, name="Alice", age=30] [id=2, name="Bob", age=25] [id=3, name="Carol", age=35]
columnar: id: [1, 2, 3]
name: ["Alice", "Bob", "Carol"]
age: [30, 25, 35]
With columns stored contiguously, a 64-byte cache line holds eight useful int64 values instead of one row's mixed fields, and SIMD instructions can process a run of values at once.
The database literature had already shown what this buys: the MonetDB/X100 work (CIDR 2005) argued that pushing batches of around a thousand column values through each operator, instead of one row at a time, is an order of magnitude or two faster. Arrow's batch-oriented design was built around exactly that model.
---
Boundary one: the driver (2018)
The first door was the smallest. DataFrame.toPandas() collected results to the driver by pickling every row and rebuilding each one as Python objects before pandas could wrap them, and it was notoriously slow. McKinney's ticket, filed on February 28, 2016, three weeks after the first commit to the Arrow repository, asked to replace that path. It stayed open for over a year. Bryan Cutler, at IBM's Spark Technology Center, resolved it on June 23, 2017, and it reached users in Spark 2.3.0 in February 2018.
Two years, for one method.
!diagram 3
# Spark 2.3 name, later renamed; on by default since Spark 4.2
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
The reverse direction, createDataFrame(pandas_df), got the same treatment (SPARK-20791). What made this matter wasn't the one method; it was the plumbing Cutler built to do it (Arrow converters on the JVM side, pyarrow integration on the Python side), which every later boundary reused.
What it removed: pickle for collecting data to the driver. What it left: everything that happens on executors. UDFs still crossed row by row.
Boundary two: UDFs, in batches (2018)
The second door was the one that mattered for performance, and it came from outside Databricks. The proposal, SPARK-21190 ("SPIP: Vectorized UDFs in Python"), was filed by Reynold Xin, but most of the implementation was done by Li Jin, an engineer at the quantitative investment firm Two Sigma and an Arrow committer, on top of Cutler's serializer. It shipped in the same Spark 2.3.0 release. Instead of calling your function once per row, Spark would send a batch of rows (10,000 by default, spark.sql.execution.arrow.maxRecordsPerBatch) as an Arrow RecordBatch and call your function once per batch, with a pandas Series:
# Before: once per row, pickle both ways
@udf(returnType="double")
def add_tax(price):
return price * 1.1 if price else None
# After: once per batch, a pandas Series in and out
@pandas_udf("double")
def add_tax(price: pd.Series) -> pd.Series:
return price * 1.1
Li Jin's post introducing the feature in October 2017 reported speedups from around 3x on simple arithmetic to around 100x on a cumulative-probability function, and the spread is the point: the simpler the function, the more of its cost was the crossing, and the more complex the function, the more it also benefits from running as vectorized NumPy instead of Python bytecode.
My own measurements, eight years and many releases later, land in the same shape. At 50 million rows, a pandas UDF did the arithmetic workload in 0.79 seconds against 5.20 for a pickled UDF, and a normal CDF in 0.82 against 6.18; on a string transform the gap was narrower, 2.51 against 6.22, for a reason that turns out to matter later.
What it removed: per-row serialization and per-row function calls. What it left: a rewrite (your function had to speak pandas), and a conversion. Every batch arrived as Arrow and was turned into pandas before your code saw it, and the result was turned back:
!diagram 4
What actually crosses the socket
Every boundary from here on sends the same thing over the socket: Arrow's IPC stream format, a schema message followed by record-batch messages, each framed like this:
┌─────────────────────────────────────────────────────────────┐
│ Continuation indicator: 0xFFFFFFFF (4 bytes) │
│ Metadata length (4 bytes, little-endian int32) │
│ Flatbuffer metadata (Message type, schema or batch header) │
│ Padding to 8-byte boundary │
│ Message body (data buffers, padded to 8-byte multiple) │
└─────────────────────────────────────────────────────────────┘
Inside the body, each column is a set of plain buffers: a validity bitmap with one bit per value, the values themselves as a contiguous typed array, and, for variable-length types like strings, an offsets array into one contiguous UTF-8 buffer. For four rows of (name: string, age: int32):
Validity bitmap (age): [1, 1, 0, 1] → 0b1011 (bit-packed)
Data buffer (age): [30, 25, 0, 35] → 16 bytes (4 × int32)
Offsets (name): [0, 5, 8, 13, 18] → 5 × int32
Data buffer (name): "AliceBobCarolDavid" (contiguous UTF-8)
No per-value type tags, no escaping, no framing between values.
Strings are where Arrow and pandas differ most: in Arrow, a million strings are one buffer and an array of offsets, while in traditional pandas they're a million separate Python objects.
---
Boundary three: letting code see Arrow (2022 to 2025)
For four years, every batch API in PySpark handed you pandas, whether you wanted pandas or not.
The first API that let user code see Arrow directly was mapInArrow, added by Hyukjin Kwon, a long-time PySpark maintainer at Databricks (SPARK-37227, with the Python side in SPARK-37228), in Spark 3.3.0 in June 2022. It passes whole partitions as an iterator of pyarrow.RecordBatch objects, and your function yields RecordBatches back:
import pyarrow as pa
import pyarrow.compute as pc
def add_tax_arrow(batches):
for batch in batches:
taxed = pc.multiply(batch.column("price"), pa.scalar(1.1))
yield pa.RecordBatch.from_arrays([batch.column("id"), taxed],
names=["id", "price_with_tax"])
df.mapInArrow(add_tax_arrow, schema="id long, price_with_tax double")
Grouped operations were the obvious remaining gap, and Enrico Minack closed it with applyInArrow (SPARK-40559), which shipped in Spark 4.0.0 in May 2025. Where applyInPandas turns each group into a pandas DataFrame, applyInArrow hands you the group as a pyarrow.Table:
def normalize(table: pa.Table) -> pa.Table:
totals = table.column("total")
mean, std = pc.mean(totals).as_py(), pc.stddev(totals).as_py()
normalized = (pa.array([0.0] * len(totals)) if not std
else pc.divide(pc.subtract(totals, pa.scalar(mean)), pa.scalar(std)))
return pa.Table.from_arrays([table.column("order_id"), totals, normalized],
names=["order_id", "total", "normalized"])
orders.groupBy("location_id").applyInArrow(
normalize, schema="order_id long, total double, normalized double")
The same idea reached the other direction in 4.0 too: the new Python Data Source API (SPARK-44076) lets a custom source written in Python yield pyarrow.RecordBatch objects instead of tuples (SPARK-48493), so data read by Python code can enter Spark as Arrow instead of being built up one Python tuple at a time.
I haven't benchmarked the grouped or data-source paths myself, so there are no numbers for them here; the mechanism is the same as everywhere else, and the benefit should be largest where the pandas conversion is most expensive, which is strings.
What it removed: the forced detour through pandas, for anyone willing to write Arrow code. What it left: that condition. Most PySpark code is plain @udf functions that nobody is going to rewrite.
Boundary four: the network (2023)
Spark Connect started as a proposal from Martin Grund at Databricks in June 2022 (SPARK-39375): separate the client from the server, so an application talks to Spark over gRPC instead of living inside the driver's JVM. Once there's a network between you and the engine, the format results travel in stops being an optimization and becomes the protocol.
The answer was Arrow.
!diagram 5
The client sends unresolved logical plans as Protocol Buffers; the server optimizes and runs them and streams the results back as Arrow record batches over the same channel. That's what makes a JVM-free Python client possible at all: the pyspark-client package is a 1.6 MB wheel of pure Python. The same columnar format that moves data between executors and Python workers now moves data between a remote cluster and a laptop.
Boundary five: the UDFs nobody rewrote (2023 to 2026)
Every boundary so far helped people who changed their code. Arrow-optimized Python UDFs (SPARK-40307, Spark 3.5.0, September 2023) took the opposite approach: keep your plain @udf exactly as written, and swap the transport underneath it. Batches cross as Arrow; Python unpacks each batch and still calls your function once per value.
!diagram 6
Xinrong Meng's 2023 Databricks post measured about 1.6x over pickle for a single UDF and 1.9x for three chained ones on a 32 GB dataset, and my benchmark agrees in spirit: useArrow=True took the arithmetic workload from 5.20 to 2.23 seconds and the CDF from 6.18 to 2.94. Real, and cheap to get, but nothing like the pandas UDF numbers, because the per-row function call is still there.
That call is my benchmark's ceiling.
Turning this on by default took two tries. It was merged for Spark 4.0 (SPARK-48516) and reverted in March 2025 before the release; it finally shipped as the default in Spark 4.2, along with Arrow for toPandas() and for Python UDTFs, and a minimum PyArrow of 18.0. On 4.2, then, every existing @udf gets Arrow transport without anyone touching it.
It's also where I made the mistake that best illustrates how confusing this boundary is. My benchmark's "Arrow" variant of the plain UDF flipped spark.sql.execution.arrow.pyspark.enabled, which governs toPandas() and createDataFrame() and has nothing to do with UDFs. The setting that actually moves a UDF onto Arrow is spark.sql.execution.pythonUDF.arrow.enabled, or useArrow=True on the decorator. So my "Arrow" row was the pickled UDF measured twice, and its near-identical result was an accidental A/A test.
Three settings are named for Arrow, and they control three different boundaries.
What it removed: pickle, for everyone, with no code changes. What it left: the function call per row, which is the part no transport format can fix.
Boundary six: Arrow as the execution format (2025)
The last step turned Arrow from the way data travels into the way your code computes. Ruifeng Zheng's SPARK-53014, "Make Arrow UDF public," shipped in Spark 4.1.0 on December 16, 2025, almost ten years after McKinney's ticket. An Arrow UDF receives pyarrow.Array values and returns one, and it's meant to be written with pyarrow.compute, whose functions are C++ kernels operating directly on Arrow buffers:
import pyarrow as pa
import pyarrow.compute as pc
from pyspark.sql.functions import arrow_udf
@arrow_udf("double")
def add_tax(price: pa.Array) -> pa.Array:
return pc.multiply(price, pa.scalar(1.1))
@arrow_udf("string")
def clean_email(email: pa.Array) -> pa.Array:
return pc.utf8_trim(pc.utf8_lower(email))
The four UDF types differ only in where the data changes format:
| UDF type | Data path | Format conversions |
|---|---|---|
| Pickle @udf | JVM → pickle → Python value → pickle → JVM | pickle both ways, per row |
| Arrow-optimized @udf | JVM → Arrow → Python value → Arrow → JVM | Arrow ↔ Python objects, per row |
| @pandas_udf | JVM → Arrow → pandas → Arrow → JVM | Arrow ↔ pandas, per batch |
| @arrow_udf | JVM → Arrow → your function → Arrow → JVM | none on the Python side |
In my benchmark, @arrow_udf was the fastest Python option everywhere, but the margin over pandas UDFs depended almost entirely on the data type. On arithmetic and the CDF it was only 5 to 9% faster; on the string transform it was 1.6x faster at both 50M and 100M rows (1.55 seconds against 2.51 at 50M).
The split comes down to strings. A numeric pandas Series is a NumPy array underneath, so converting from Arrow is cheap; a string column in traditional pandas is an array of Python objects, so converting means building millions of them. pyarrow.compute skips all of that and works on the one contiguous buffer.
What it removed: the last conversion on the Python side. What it left: Spark's side.
Taken together, the choice today comes down to what your function can receive:
!diagram 7
The pipe itself
Everything above changes what goes through the socket. Hyukjin Kwon's SPARK-51688, shipped in Spark 4.1.0, changes the socket: instead of TCP over loopback, which runs every byte through the full TCP/IP stack even though it never leaves the machine, the JVM and Python workers can talk over a Unix domain socket, which goes straight through a kernel buffer.
It's opt-in (spark.python.unix.domain.socket.enabled, whose default comes from the PYSPARK_UDS_MODE environment variable) in 4.1 and 4.2, and that's unchanged on the 4.3 branch, which instead adds another transport option: pipelined, asynchronous transfer of UDF batches between the JVM and Python (spark.python.udf.pipelined.enabled, off by default).
!diagram 8
Spark 1.x: pickle format + TCP loopback slow format over a slow pipe
Spark 2.3: Arrow format + TCP loopback fast format over a slow pipe
Spark 4.1: Arrow format + Unix domain socket fast format over a faster pipe (opt-in)
---
One boundary at a time
!diagram 9
As of Spark 4.2, Arrow is the format at every place data crosses between the JVM and Python:
| Boundary | Arrow's role | Default |
|---|---|---|
| toPandas() / createDataFrame() | replaces pickle for driver data collection | yes, since 4.2 |
| @udf (plain Python UDFs) | replaces pickle transport | yes, since 4.2 |
| @pandas_udf, mapInPandas, applyInPandas | batch transport, converted to pandas | always |
| @arrow_udf, mapInArrow, applyInArrow | transport and execution format | always |
| Spark Connect results | the result protocol | always |
| Python Data Source API | source data format | when the source yields RecordBatches |
The last boundary
What's left is on Spark's side of the wall. Inside the JVM, Spark still executes on its own row format (UnsafeRow, from the Tungsten project), and Arrow exists at the edges, so every crossing still begins and ends with a conversion between Spark's rows and Arrow's columns. Engines like Velox and DataFusion show that a query engine can execute on Arrow-format data directly; Spark doesn't, and closing that gap is a much bigger change than any of the boundaries above.
!diagram 10
The 4.3 branch, which hasn't been released yet (its first release candidate failed its vote on September 6, and a second, put to a vote on October 13, drew -1 votes over critical fixes that still need backporting, so there's no date), has two small signs of movement at exactly this edge. One is an Arrow format for Spark's in-memory cache (ArrowCachedBatchSerializer, SPARK-57268), part of a proposal to make Spark better on a single machine: the first time Arrow shows up as a storage format inside Spark itself rather than at a door.
The other goes around the boundary instead of through it: experimental transpilation of Python UDFs into Spark expressions (SPIP SPARK-54783), so a simple enough Python function never has to cross into Python at all. It's off by default, classic-mode only, and requires ANSI mode, but it's the one approach that could get under my benchmark's ceiling, because it removes the per-row Python call rather than making the trip to it cheaper.
The pandas side is moving too, more slowly than you'd expect. pandas 3.0, released in January 2026, made a PyArrow-backed string type the default, which should shrink exactly the string-conversion cost that separates pandas UDFs from Arrow UDFs. But PySpark doesn't support pandas 3 yet, not even on the 4.3 branch: it still pins pandas below 3.0, and the umbrella ticket for supporting it (SPARK-55139) is open.
When Wes McKinney filed SPARK-13534 in 2016, he was asking for one method to stop pickling. Ten years later, every door he could have named uses Arrow, the defaults finally say so, and the conversion that remains is the one Spark performs on itself, between its own rows and everything outside it.
That's the last boundary, and it was never a Python problem.
1 reaction
Most PySpark Tuning Advice Starts in the Wrong Place
When I benchmarked PySpark's UDF types earlier this year, I ran the same three workloads (trivial arithmetic, a string transform, a normal CDF) at five scales, from 100 thousand rows to 100 million. At 100 million rows, the default row-at-a-time Python UDF took about 10 seconds for arithmetic that a built-in function finished in under a fifth of a second. At 100 thousand rows, the same UDF was the fastest Python option I tested.
Both results are true, and between them they're most of what you need to know about PySpark performance: the cost is almost never in your function. It's in how many times, and in what shape, your data crosses from the JVM into Python, and whether that crossing is large enough to matter at all.
My argument is that there is really one PySpark performance technique, applied at different places: cross the JVM/Python boundary less often, move less across it, or stop paying for a JVM where you don't need one. The changes worth making are ordered by how much of that crossing they remove, and the biggest one, not crossing at all, is also the most often skipped.
The history of how Arrow came to carry that crossing is in Arrow in PySpark, 2016 to 2026, and the full benchmark is in Stop Blaming Serialization for Slow Python UDFs. This post is the practical version: what to change, in what order, and what each change actually removes.
---
Where the time goes
PySpark is a Python client for a JVM engine. DataFrame operations build a plan in Python (through Py4J, which only ever carries control messages, never bulk data) and the JVM executes it; df.filter(col("x") > 5).groupBy("y").count() never puts a row into Python, which is why it runs at the same speed as Scala. The data only leaves the JVM when your own code needs it:
!diagram 1
Everything outside arrows ① and ② happens at native speed. Every change below makes those two arrows smaller, rarer, or unnecessary, and I've ordered them by how much they took off the arrows in what I measured, or, where I haven't measured, by what they remove.
Don't cross at all: built-ins and SQL UDFs
The largest win isn't a better UDF; it's no UDF. Built-in functions run inside the JVM as Catalyst expressions, fused by whole-stage code generation into tight loops, and never serialize anything.
What surprised me more is that SQL UDFs get the same treatment. Spark 4.0 added SQL user-defined functions (SPARK-46057), and their bodies are inlined into the logical plan, so the optimizer can see straight through them:
CREATE FUNCTION classify(score INT) RETURNS STRING
RETURN CASE
WHEN score > 90 THEN 'excellent'
WHEN score > 70 THEN 'good'
ELSE 'needs_improvement'
END;
In my benchmark (median wall-clock time, 50M rows, Spark 4.1.0 through Spark Connect on one machine), the SQL UDF was indistinguishable from the built-ins, and every Python option cost several times more:
| | arithmetic | string | normal CDF |
|---|---|---|---|
| built-in functions | 0.107 s | 0.486 s | 0.176 s |
| SQL UDF | 0.095 s | 0.477 s | n/a |
| fastest Python UDF (@arrow_udf) | 0.725 s | 1.545 s | 0.762 s |
| default Python UDF (@udf, pickle) | 5.199 s | 6.216 s | 6.183 s |
So the first question for any UDF is whether it needs to be Python. Plenty of "we need a UDF" logic is a CASE expression, a string function, or a date calculation that already exists in pyspark.sql.functions. Python earns its place when you need a library (scikit-learn, a parser, a C extension) or logic that genuinely can't be expressed in SQL. Everything after this point is about paying less for that choice.
Cross once per batch, not once per row
If it has to be Python, the next biggest lever is how often Spark calls your function. A regular @udf is called once per row; @pandas_udf and @arrow_udf are called once per batch of rows. That difference showed up as a hard ceiling in my runs: every row-at-a-time UDF, whatever it computed, topped out at roughly 8 to 10 million rows per second, which is roughly the rate at which CPython can be entered and exited. The batched UDFs weren't bound by it.
| 50M rows | arithmetic | string | normal CDF |
|---|---|---|---|
| @udf (pickle, per row) | 5.199 s | 6.216 s | 6.183 s |
| @udf(useArrow=True) (Arrow transport, still per row) | 2.228 s | 4.430 s | 2.942 s |
| @pandas_udf (per batch, pandas.Series) | 0.789 s | 2.512 s | 0.819 s |
| @arrow_udf (per batch, pyarrow.Array) | 0.725 s | 1.545 s | 0.762 s |
Two things in that table decide what to change. Moving a per-row UDF onto Arrow transport helps, 1.4x to 2.4x here, and on Spark 4.2 you get it for free: 4.2 finally flipped spark.sql.execution.pythonUDF.arrow.enabled (and its UDTF counterpart) to true by default, after the same change was merged for 4.0 and reverted before release (SPARK-48516).
But rewriting to a batched UDF is worth several times more again, because it removes the per-row call rather than making the trip to it cheaper.
Between the two batched options, the choice mostly comes down to strings: @arrow_udf was only 5 to 9% faster than @pandas_udf on numeric work, but about 1.6x faster on the string transform, because pandas has traditionally stored strings as individual Python objects while Arrow keeps them in one contiguous buffer.
import pyarrow as pa
import pyarrow.compute as pc
from pyspark.sql.functions import arrow_udf
@arrow_udf("string")
def add_suffix(x: pa.Array) -> pa.Array:
return pc.binary_join_element_wise(pc.utf8_upper(x), pa.scalar("_SUFFIX"), "")
One warning, because I learned it the expensive way: spark.sql.execution.arrow.pyspark.enabled is not the UDF setting. It governs toPandas() and createDataFrame(), and it's the one my benchmark originally toggled for its "Arrow" case, which is why I once reported that Arrow transport changes nothing for UDFs. It was the same pickled UDF, measured twice. (On 4.2 that setting is also on by default, so the confusion matters less for new code than for anyone reading old tuning advice.)
And the scale caveat from the opening is real. At small sizes, fixed costs dominate, and the ordering can flip:
| arithmetic, median seconds | built-in | @pandas_udf | @udf(useArrow=True) | @udf (pickle) |
|---|---|---|---|---|
| 100K rows | 0.073 | 0.181 | 0.212 | 0.161 |
| 10M rows | 0.074 | 0.299 | 0.568 | 1.092 |
| 100M rows | 0.147 | 1.400 | 4.188 | 10.139 |
If your job touches a few hundred thousand rows, rewriting its UDFs is unlikely to be what makes it faster.
Don't produce the data: Python data sources with pushdown
After calling Python less, the next thing to remove is data that never needed to exist. Built-in sources like Parquet already push filters down and skip row groups that can't match; for most of Spark's history, custom sources that could do the same had to be written in Java or Scala. Spark 4.0 added the Python Data Source API (SPARK-44076), and Spark 4.1 added filter pushdown for it (SPARK-51271), so a source written in Python can now turn WHERE status = 'active' into a query parameter, a database predicate, or a skipped file block, and never send the rest.
When I went back over my own notes on this feature to write this post, the example I'd written was wrong, and checking it against Spark's source turned up something I hadn't known: pushdown for Python sources is off by default. It's governed by spark.sql.python.filterPushdown.enabled, an internal setting added in 4.1 whose description says enabling it costs "running Python worker one additional time during planning"; without it, your pushFilters() is never called.
The method also has a less obvious contract than I'd assumed. It returns the filters Spark must still apply after the scan, not the ones you accepted, and each filter's attribute is a column-path tuple:
from pyspark.sql.datasource import DataSourceReader, EqualTo
spark.conf.set("spark.sql.python.filterPushdown.enabled", "true") # off by default
class ApiReader(DataSourceReader):
def __init__(self, options):
self.url = options.get("url")
self.pushed = []
def pushFilters(self, filters):
# Filters arrive ANDed together; yield the ones Spark must still apply.
for f in filters:
if isinstance(f, EqualTo) and f.attribute == ("status",):
self.pushed.append(f) # handled at the source
else:
yield f
def read(self, partition):
import requests
params = {f.attribute[0]: f.value for f in self.pushed}
for row in requests.get(self.url, params=params).json():
yield (row["id"], row["name"], row["status"])
Only conjunctions are pushed (a filter like a = 1 or b = 2 arrives as an empty list), and a reader can also yield pyarrow.RecordBatch objects instead of tuples, which the Spark docs say can improve performance by "up to one order of magnitude."
I haven't measured either path, so this is the one change on the list I'm ranking by what it removes rather than by what I saw: for a source backed by an API or a database, rows that are never fetched are the cheapest rows there are.
Skip pandas for grouped work
Per-group logic (normalize within each location, fit a model per customer) has long meant applyInPandas, which converts every group from Arrow into a pandas DataFrame and back. Spark 4.0 added applyInArrow (SPARK-40559), and each group now reaches your function as a pyarrow.Table, in the format it crossed the boundary in:
def normalize(table: pa.Table) -> pa.Table:
totals = table.column("total")
mean, std = pc.mean(totals).as_py(), pc.stddev(totals).as_py()
normalized = (pa.array([0.0] * len(totals)) if not std
else pc.divide(pc.subtract(totals, mean), std))
return pa.Table.from_arrays([table.column("order_id"), totals, normalized],
names=["order_id", "total", "normalized"])
result = orders.groupBy("location_id").applyInArrow(
normalize, schema="order_id long, total double, normalized double")
This is the same trade as step 2, one level up: it removes a conversion, not a crossing, so the payoff should track what that conversion cost in my UDF numbers (little for numeric groups, more for string-heavy ones, more for larger groups). I haven't benchmarked the grouped APIs, so I won't put a number on it.
Take the JVM out of the client
Spark Connect is on this list for a different reason than the others. It doesn't make a query faster; it removes the JVM from every client process. Classic PySpark embeds a full Spark driver in each notebook, scheduled job, and CI container. With Connect, the client sends unresolved plans over gRPC and gets Arrow batches back, and the thin pyspark-client package has no JVM and needs no JDK. When I measured it for my own Connect demos on 4.2, the numbers were less dramatic than the pitch:
| | pyspark (full) | pyspark-client |
|---|---|---|
| PyPI artifact | 429.28 MB | 1.60 MB |
| installed pyspark/ | 481 MB | 19 MB |
| Spark JARs on the client | 456 MB, 276 files | none |
| virtual env with Connect dependencies | 820 MB | 359 MB |
The last row is the honest one: pyarrow, pandas, and numpy are required dependencies, so the environment shrinks by less than half.
What you actually get is a client with no JVM to start, no JDK to install, and no way for a crashed notebook to take the driver down with it. What you give up is anything that needs the JVM on your side: RDDs, SparkContext, custom accumulators. ML on Connect has been supported since 4.0 and generally available since 4.1, so that's no longer on the list of reasons to stay classic.
Give Python the memory it actually uses
The last change is the one that turns slow jobs into failed ones when it's wrong. Python workers live outside the JVM heap, so the settings most Spark tuning advice is about don't cover them:
!diagram 2
The rule is that the container has to fit the heap, the overhead, and the OS, and that Python's share has to come out of the overhead. If you use toPandas() or pandas UDFs, spark.sql.execution.arrow.pyspark.selfDestruct.enabled frees Arrow buffers as they're converted, so the same data doesn't sit in memory twice.
Beyond that, the usual advice still holds: let Adaptive Query Execution size shuffle partitions at runtime (it's been on by default since 3.2, and 4.1 extended it to stateless streaming, SPARK-53941, which can change a streaming query's behaviour after an upgrade), and broadcast small dimension tables instead of shuffling them. None of that is specific to Python, which is why it's last on a list about Python.
---
What's coming in 4.3
The 4.3 branch hasn't been released (its first release candidate failed its vote on September 6, and there's no new date), but it shows where the boundary is heading, and it confirms the order above.
The most interesting change goes around the boundary instead of through it: experimental transpilation of Python UDFs into Spark expressions (SPIP SPARK-54783, spark.sql.experimental.optimizer.transpilePyUDFs). A simple enough Python function would never cross into Python at all, which is change number 1 done for you. It's off by default, requires ANSI mode, and only works in classic mode, not Connect, but it's the only approach on the branch that could get under the per-row ceiling rather than lowering the cost of hitting it.
Alongside it are smaller changes to the crossing itself: pipelined, asynchronous transfer of UDF batches (spark.python.udf.pipelined.enabled, SPARK-56642, off by default) and a cap on bytes per UDF batch (SPARK-57593). A proposal to make Spark better on a single machine also adds an Arrow format for Spark's in-memory cache (SPARK-57268).
Two things haven't moved. PySpark still doesn't support pandas 3, even on the 4.3 branch; it pins pandas below 3.0, and the ticket to support it (SPARK-55139) is open, so the PyArrow-backed string type pandas 3 made the default in January won't help pandas UDFs yet. And Real-Time Mode in 4.3 gains stateful operators, but still can't run Python UDFs (SPARK-57237 is open).
Housekeeping to plan for: 4.2 raised the minimum PyArrow to 18.0.0 and dropped PyPy, and 4.3 drops Python 3.10 (SPARK-56889).
---
I opened with a UDF that took 10 seconds at 100 million rows and was the fastest Python option at 100 thousand. The reason is the same one behind every change on this list: your function was never the expensive part.
The crossing was, and the best PySpark optimization is still the one that decides the data doesn't need to cross.
1 reaction
Apache Spark won't melt down when your data doesn't fit in RAM anymore
Hey, before I start I need to make an honest clarification, because in this series transparency is sacred: the automated analysis I was handed for this post talked about a "memory layer for AI agents," vector DBs, and cross-session context persistence. That has nothing to do with what Apache Spark actually is. That description belongs to a different tool — probably something like Mem0 or similar — and it got crossed wrong in the curation pipeline. It happens, automated systems sometimes mix up metadata. That's why the human verdict still says "pending" — here it is, done by hand, with the real data.
What I do have confirmed: Apache Spark shows up in 3 independent awesome lists, and the official repo description describes it as a framework with "micro-batch processing for streams" and "stateful exactly-once semantics" as a backend. In plain terms: it's a distributed engine built for workloads where a single server, no matter how much RAM you throw at it, becomes a bottleneck.
Picture this: you have logs from an app generating 500 GB per day, and you need to calculate aggregate metrics — unique users, average latencies, anomaly detection — without waiting 14 hours for a Python script running on a single core to finish. That's where Spark stops being an academic curiosity and becomes the tool that saves your sprint.
What it does
Apache Spark is a distributed data processing engine, originally written in Scala, that runs on the JVM (yes, the JVM again — there's a reason I keep mentioning it in this series). It was born in Berkeley's AMPLab as a direct response to Hadoop MapReduce's limitations: Spark processes in memory instead of writing to disk at every intermediate step, which makes it orders of magnitude faster for iterative workloads (think ML algorithms, which need to pass over the same data again and again).
The central concept is the RDD (Resilient Distributed Dataset): a collection of data partitioned across the cluster's nodes, immutable, and able to rebuild itself if a node goes down (hence "resilient"). On top of that sit friendlier layers like DataFrames and Datasets, which give you a SQL/Pandas-style API, but distributed.
// Classic example: counting words in a giant dataset
// distributed across all nodes in the cluster
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder()
.appName("ContadorDePalabras")
.getOrCreate()
val textFile = spark.read.textFile("hdfs://logs/*.txt")
val conteo = textFile
.flatMap(linea => linea.split(" ")) // split into words
.groupByKey(identity) // group by word
.count() // count occurrences
conteo.show()
spark.stop()
The interesting part is that this exact same code runs the same way on your notebook with 4 GB of RAM as it does on a 200-node cluster on AWS EMR. Spark takes care of partitioning the data, distributing the tasks, and handling failures — you just describe what you want to do, not how to distribute it.
The streaming part (the one the official repo description mentioned, "micro-batch processing for streams... stateful exactly-once semantics") is Spark Structured Streaming: instead of processing event by event like Kafka Streams or Flink, Spark groups events into micro-batches (every few seconds) and processes them with the same API you use for batch. That gives you "exactly-once" — no event gets processed twice or lost, even if a node explodes at the worst possible moment.
# Streaming in PySpark: counting events by time window
# reading from a Kafka topic in real time
df = spark.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "localhost:9092") \
.option("subscribe", "eventos") \
.load()
conteoPorVentana = df \
.groupBy(window(df.timestamp, "1 minute")) \
.count()
query = conteoPorVentana.writeStream \
.outputMode("update") \
.format("console") \
.start()
query.awaitTermination()
The project is open source under the Apache 2.0 license, maintained by the Apache Software Foundation, with official support for Scala, Java, Python (PySpark), and R. The repo is at apache/spark.
Why it's on the list
Showing up in 3 independent awesome lists is a decent signal — different curators, different angles, pointing at the same tool. It doesn't prove consensus across entire industries, but it's more than "trending on Twitter": it suggests the tool holds up across more than one context.
What sets it apart from alternatives like Hadoop MapReduce (its direct predecessor, much slower because it writes to disk at every step) or Dask (more pythonic but with an ecosystem and maturity well below) is the combination of three things: in-memory speed, a unified API that serves both batch and streaming, and an ecosystem of integrated libraries (MLlib for distributed machine learning, GraphX for graphs, Spark SQL for database-style queries) that saves you from having to glue together five different tools.
If you're coming from my history with infrastructure — Linux, networking, servers — you'll quickly understand why Spark earns my respect: it's not magic, it's distributed systems engineering done right, with two decades of academic papers behind it (the original RDD paper by Zaharia et al. is required reading if you like understanding the why, not just the what).
When NOT to use it
Now for the honest part, because in this series we don't sell smoke: if your dataset fits comfortably in a single machine's RAM (say, under 10-20 GB depending on the hardware), setting up a Spark cluster is like using a crane to lift a soda can. Pandas, Polars, or even DuckDB will give you faster results with a fraction of the operational complexity.
It's also not the best option if you need real sub-second latency in streaming — there, Apache Flink beats it hands down because it processes event by event, not in micro-batches. And if your team doesn't have experience operating distributed clusters (YARN, Kubernetes, or Databricks as a managed layer), the maintenance cost can eat up any performance gain. Spark isn't "install and go": it requires understanding partitioning, shuffle, executor memory — things you learn the hard way.
Wrap-up
This is post #13 of "Awesome Curated: The Tools," the series where we break down tools that passed the filter of our curation system — cross-referencing awesome lists, AI analysis, and a final human verdict. Spark is one of those tools that doesn't have the shine of something new, but that's still there, carrying the real weight of the industry, long after the hype moved on to something else.
If you're interested in the infrastructure and systems side of things, check out the post on Node.js and the runtime that changed the backend or the one on Sniffnet for monitoring your network. And if you want to see the full arc of the series, the complete list is at /blog/series/awesome-curated-tools.
---
This article was originally published on juanchi.dev
1 reaction · 1 comment
Mentions
Пишем высокопроизводительный rate-limiter на golang О чём статья Недавно я познакомился с golang и в качестве практики решил реализовать алгоритмы rate limiting'а из статьи c arxiv. В этой статье мы рассмотрим написание rate-limiter'а в виде отдельно стоящего сервиса и в виде middleware, используя in-memory кэш или Redis, добавим поддержку REST и gRPC, подключим Prometheus и поймём как добиться атомарности операций с минимальными задержками. Кому полезно Новичкам в golang или же людям, которые по какой-то причине также решили использовать собственные реализации rate-limiter'ов заместо уже существующих решений. Предисловие Указанная ранее статья покрывает 3 алгоритма rate-limiting'а: Token bucket Fixed window Rolling window Суть данных алгоритмов вы можете прочитать самостоятельно, я же сфокусируюсь на том, как написать данные алгоритмы таким образом, чтобы rate-limiter работал быстро и надёжно. Конфигурация Первое, с чего стоит начать, это конфигурация вашего limiter'а. В своей реализации я ставил следующие требования: Возможность переключаться между алгоритмами Поддержка in-memory кэша и Redis Метрики в Prometheus Поддержка REST и gRPC Из этих требований вытекает следующий файл конфигурации: store: "redis" use_algo: "token_bucket" metrics: use_prometheus: true port: 9091 algo_settings: capacity: 10 rate: 0.1 backends: in_memory: default_ttl: 30 # seconds eviction_time: 120 # seconds shards_count: 256 redis: addr: "redis:6379" password: "m1re1di77sss" db: 0 default_ttl: 1000sourcecraft.dev Рассмотрим его подробнее: store отвечает за используемое хранилище, может быть "redis" или же "in_memory" use_algo отвечает за используемый алгоритм metrics позволяет включать и отключать поддержку выгрузки и подсчёта метрик во время работы, а также за порт, на котором позже будет поднят мультиплексер для метрик algo_settings для каждого алгоритма разные, отвечает за настройки, которые будут использоваться алгоритмом backends - настройки каждого из хранилищ, в in_memory default_ttl отвечает за ttl ключей в хранилище, eviction_time отвечает за частоту очистки in-memory хранилища от протухших ключей, shard_count отвечает за то, на сколько частей in-memory кэш будет шардирован, число должно быть степенью 2, позже я объясню почему Конфигурация для middleware выглядит похожим образом: type InMemoryBackendSettings struct { DefaultTtl int64 yaml:"default_ttl" EvictionTime int64 yaml:"eviction_time" ShardsCount uint32 yaml:"shards_count" } type RedisBackendSettings struct { Addr string yaml:"addr" Password string yaml:"password" Db int yaml:"db" DefaultTtl int64 yaml:"default_ttl" } type AlgoSettings struct { Capacity *float64 yaml:"capacity" Rate *float64 yaml:"rate" WindowSize *int64 yaml:"window_size" MaxRequests *int64 yaml:"max_requests" }sourcecraft.dev Бэкенды для limiter'а Думаю, ни для кого не секрет, что Redis умеет выполнять Lua скрипты атомарно, что является идеальной функциональной возможностью для надёжного rate-limiting'а. В таком случае наш бэкенд limiter'а также должен уметь выполнять операции атомарно, отсюда вытекает интерфейс бэкенда: type InMemoryStoreInterface interface { Update(key string, newFn func() any, fn func(v any) error) error Close() error }sourcecraft.dev Update принимает на вход ключ, по которому мы хотим принять решение, а также 2 колбэка: newFn - функция создания нового объекта в хранилище, в случае если данный ключ в хранилище ещё не появлялся. Это позволяет избежать создания данного объекта в случае, если ключ уже содержится в хранилище. Конечно, можно было сначала проверять и только потом добавлять, но это добавляет лишней логики клиенту, использующему хранилище. fn - собственно функция, содержащая логику limiting'а In-memory кэш Наше хранилище под капотом будет использовать мапу, во избежание race condition'ов мы будем использовать mutex. Но ставить mutex на всю мапу слишком дорого, так как другие клиенты всё ещё должны пользоваться хранилищем. Отсюда вытекает необходимость в шардировании. Пока одна горутина выполняет логику по ключу x, мы всё ещё можем работать с ключом y: type shard struct { mu sync.Mutex entries map[string]*cacheEntry } type cacheEntry struct { entry any lastAccess int64 } type ShardedInMemoryStore struct { shards []*shard shardsCount uint32 expirationTime int64 ticker *time.Ticker done chan any once sync.Once } func NewCache( expirationTime int64, evictionInterval time.Duration, shardCount uint32, ) *ShardedInMemoryStore { ticker := time.NewTicker(evictionInterval) cache := &ShardedInMemoryStore{ shards: make([]*shard, shardCount), shardsCount: shardCount, expirationTime: expirationTime, done: make(chan any), ticker: ticker, } for i := range shardCount { cache.shards[i] = &shard{entries: make(map[string]*cacheEntry)} } go func() { defer ticker.Stop() for { select { case <-cache.done: return case <-ticker.C: cache.handleEviction() } } }() return cache } sourcecraft.dev Также удаление протухших ключей теперь не будет лочить всю мапу, а только один конкретный шард: func (cache *ShardedInMemoryStore) handleEviction() { currentTime := time.Now().Unix() for _, shard := range cache.shards { shard.mu.Lock() for key, value := range shard.entries { if currentTime-cache.expirationTime > value.lastAccess { delete(shard.entries, key) } } shard.mu.Unlock() } }sourcecraft.dev А также ранее упомянутый Update: func (cache *ShardedInMemoryStore) Update(key string, createFn func() any, fn func(v any) error) error { shard := cache.getShard(key) shard.mu.Lock() defer shard.mu.Unlock() now := time.Now().Unix() value, present := shard.entries[key] if !present { value = &cacheEntry{entry: createFn(), lastAccess: now} shard.entries[key] = value } value.lastAccess = now return fn(value.entry) }sourcecraft.dev Именно здесь реализуется столь необходимая атомарность, мы лочим mutex на шарде и выполняем операцию, не блокируя другие шарды и не допуская race condition'ов. Так почему количество шардов должно быть степенью двойки? Для того, чтобы распределять клиентские ключи более-менее равномерно по шардам нужно использоват функцию хэширования, которая не только будет раскладывать ключи равномерно, но и будет быстрой. Под такую функцию подходит Fowler–Noll–Vo. А чтобы после этого быстро замапить хэш на шард можно воспользоваться трюком. Вкратце можно очень быстро искать остатки от деления на степень двойки просто используя формулу: hash & (shardsCount - 1): func (cache *ShardedInMemoryStore) getShard(key string) *shard { h := uint32(2166136261) for i := 0; i < len(key); i++ { h ^= uint32(key[i]) h *= 16777619 } return cache.shards[h&(cache.shardsCount-1)] }sourcecraft.dev Redis кэш С Redis всё проще: package cache import ( "context" "github.com/redis/go-redis/v9" ) type RedisStoreInterface interface { RunLuaScript(ctx context.Context, script *redis.Script, keys []string, args ...any) (bool, int64, error) DefaultTtl() int64 }sourcecraft.dev Передаём контекст, скрипт и массив ключей, в данном случае массив ключей это требование библиотеки для работы с Redis в Golang. Массив всегда будет состоять из одного элемента. DefaultTtl - это getter, который используется для token bucket алгоритма, чтобы удалением ключа при использовании Redis занимался сам Redis, а не приложение. package cache import ( "context" "fmt" "github.com/redis/go-redis/v9" ) type RedisStore struct { client *redis.Client defaultTtl int64 } func NewRedisStore(client *redis.Client, defaultTtl int64) *RedisStore { return &RedisStore{client: client, defaultTtl: defaultTtl} } // RunLuaScript runs a Lua script that returns {allowed(0|1), retryAfter}. func (r *RedisStore) RunLuaScript( ctx context.Context, script *redis.Script, keys []string, args ...any, ) (bool, int64, error) { res, err := script.Run(ctx, r.client, keys, args...).Int64Slice() if err != nil { return false, 0, err } if len(res) != 2 { return false, 0, fmt.Errorf("unexpected length of script output") } return res[0] == 1, res[1], nil } func (r *RedisStore) DefaultTtl() int64 { return r.defaultTtl }sourcecraft.dev Используем созданные хранилища Теперь рассмотрим функцию запуска limiter'а: package main import ( "context" "fmt" "log" "net" "net/http" "os" "os/signal" "qwrttqr-rate-limiter/core/config" "qwrttqr-rate-limiter/core/server/algos" "qwrttqr-rate-limiter/core/server/facade" "qwrttqr-rate-limiter/core/server/limiting" "qwrttqr-rate-limiter/core/server/metrics" ratelimiterv1 "qwrttqr-rate-limiter/gen/ratelimiter/v1" "strconv" "syscall" "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" "google.golang.org/grpc" ) func main() { if err := run(); err != nil { log.Fatal(err) } } func run() error { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() cfg, err := config.ReadConfig() if err != nil { return fmt.Errorf("failed to read config: %w", err) } rl, err := limiting.NewRateLimiter(cfg) if err != nil { return fmt.Errorf("create rate limiter: %w", err) } defer rl.Close() var limiter algos.Limiter = rl errCh := make(chan error, 3) shutdownMetrics := func(ctx context.Context) {} if cfg.Metrics.UsePrometheus { reg := prometheus.NewRegistry() rec := metrics.NewPromRecorder(reg) limiter = metrics.Instrument(limiter, rec) metricsMux := http.NewServeMux() metricsMux.Handle("GET /metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{})) metricsSrv := &http.Server{Addr: ":" + strconv.Itoa(cfg.Metrics.Port), Handler: metricsMux, ReadHeaderTimeout: 5 * time.Second} go func() { errCh <- metricsSrv.ListenAndServe() }() log.Printf("metrics available on %s", metricsSrv.Addr) shutdownMetrics = func(ctx context.Context) { _ = metricsSrv.Shutdown(ctx) } } limitingMux := http.NewServeMux() limitingMux.HandleFunc("GET /health", facade.HealthHandler) limitingMux.HandleFunc("POST /limit", facade.LimitHandler(limiter)) limitingSrv := &http.Server{Addr: ":8080", Handler: limitingMux, ReadHeaderTimeout: 5 * time.Second} grpcSrv := grpc.NewServer() ratelimiterv1.RegisterRateLimiterServer(grpcSrv, facade.NewGRPCServer(limiter)) tcpLis, err := net.Listen("tcp", ":8081") if err != nil { return fmt.Errorf("listen grpc: %w", err) } go func() { errCh <- limitingSrv.ListenAndServe() }() log.Printf("listening http on %s", limitingSrv.Addr) go func() { errCh <- grpcSrv.Serve(tcpLis) }() log.Printf("listening grpc on %s", tcpLis.Addr()) select { case err := <-errCh: return err case <-ctx.Done(): } shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() grpcDone := make(chan struct{}) go func() { grpcSrv.GracefulStop(); close(grpcDone) }() select { case <-grpcDone: case <-shutdownCtx.Done(): grpcSrv.Stop() } shutdownMetrics(shutdownCtx) return limitingSrv.Shutdown(shutdownCtx) } sourcecraft.dev Давайте по порядку: ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop()sourcecraft.dev Создаём shutdown context, который срабатывает на нажатие Ctrl-C либо при получении syscall.SIGTERM (сигнал, посылаемый родительским процессом на завершение нашего процесса, например когда тушим докер-контейнер). defer stop() нужен для того, чтобы прекратить обрабатывать данные сигналы, сразу перед возвратом из run. cfg, err := config.ReadConfig() if err != nil { return fmt.Errorf("failed to read config: %w", err) } rl, err := limiting.NewRateLimiter(cfg) if err != nil { return fmt.Errorf("create rate limiter: %w", err) } defer rl.Close()sourcecraft.dev Далее читаем конфиг и создаём сам limiter, у limiter'а есть функция Close, нужная для закрытия соединения с Redis. в in-memory кэше она просто закрывает канал, что останавливает удаление ключей из кэша. var limiter algos.Limiter = rl errCh := make(chan error, 3)sourcecraft.dev Следующие строки позволяют создать указатель с типом интерфейса на limiter: package algos import ( "context" ) type Request struct { Key string RequiredTokens float64 } type Decision struct { Allowed bool // RetryAfter is seconds RetryAfter int64 } type Limiter interface { Allow(ctx context.Context, req Request) (Decision, error) }sourcecraft.dev Так как NewRateLimiter возвращает не интерфейс, а указатель на структуру, поддерживающую Close и Allow: type RateLimiter struct { algos.Limiter io.Closer } func NewRateLimiter(cfg config.StandaloneServiceConfiguration) (_ *RateLimiter, err error) { var ( mem cache.InMemoryStoreInterface rdb *redis.Client closer io.Closer ) switch cfg.Store { case "in_memory": cacheInstance := cache.NewCache( cfg.Backends.InMemory.DefaultTtl, time.Duration(cfg.Backends.InMemory.EvictionTime)*time.Second, cfg.Backends.InMemory.ShardsCount, ) mem, closer = cacheInstance, cacheInstance case "redis": redisClient := redis.NewClient(&redis.Options{ Addr: cfg.Backends.Redis.Addr, Password: cfg.Backends.Redis.Password, DB: cfg.Backends.Redis.Db, }) rdb, closer = redisClient, redisClient default: return nil, fmt.Errorf("unsupported store: %q", cfg.Store) } defer func() { if err != nil { _ = closer.Close() } }() var limiter algos.Limiter switch cfg.UsedAlgo { case "token_bucket": if err := algos.ValidateTokenBucketConfig(cfg.AlgoSettings); err != nil { return nil, err } store, err := algos.NewTokenBucketStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl) if err != nil { return nil, err } limiter = &algos.TokenBucketLimiter{ Capacity: *cfg.AlgoSettings.Capacity, Rate: *cfg.AlgoSettings.Rate, Store: store, Now: realClock, } case "fixed_window": if err := algos.ValidateFixedWindowConfig(cfg.AlgoSettings); err != nil { return nil, err } store, err := algos.NewFixedWindowStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl) if err != nil { return nil, err } limiter = &algos.FixedWindowLimiter{ MaxRequests: *cfg.AlgoSettings.MaxRequests, WindowSize: *cfg.AlgoSettings.WindowSize, Store: store, Now: realClock, } case "rolling_window": if err := algos.ValidateRollingWindowConfig(cfg.AlgoSettings); err != nil { return nil, err } store, err := algos.NewRollingWindowStore(cfg.Store, mem, rdb, cfg.Backends.Redis.DefaultTtl) if err != nil { return nil, err } limiter = &algos.RollingWindowLimiter{ MaxRequests: *cfg.AlgoSettings.MaxRequests, WindowSize: *cfg.AlgoSettings.WindowSize, Store: store, Now: realClock, } default: return nil, fmt.Errorf("unsupported rate limiting algorithm: %s", cfg.UsedAlgo) } return &RateLimiter{Limiter: limiter, Closer: closer}, nil }sourcecraft.dev Возвращаясь обратно в run: shutdownMetrics := func(ctx context.Context) {} if cfg.Metrics.UsePrometheus { reg := prometheus.NewRegistry() rec := metrics.NewPromRecorder(reg) limiter = metrics.Instrument(limiter, rec) metricsMux := http.NewServeMux() metricsMux.Handle("GET /metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{})) metricsSrv := &http.Server{Addr: ":" + strconv.Itoa(cfg.Metrics.Port), Handler: metricsMux, ReadHeaderTimeout: 5 * time.Second} go func() { errCh <- metricsSrv.ListenAndServe() }() log.Printf("metrics available on %s", metricsSrv.Addr) shutdownMetrics = func(ctx context.Context) { _ = metricsSrv.Shutdown(ctx) } }sourcecraft.dev Мы помечали rl, как структуру, реализующую интерфейс как раз ради поддержки метрик здесь. О том, как метрики работают я расскажу позже, сейчас достаточно понимания того, что мы создаём специальный HTTP сервер, с которого Prometheus будет забирать метрики. Для подсчёта данных метрик мы используем Registry, а в функции instrument мы оборачиваем limiter в мидлварь, которая будет обновлять счётчики. Функция shutdownMetrics обновляется в зависимости от того, включаем ли у себя метрики или нет, это нужно, чтобы сделать graceful shutdown данного сервера. limitingMux := http.NewServeMux() limitingMux.HandleFunc("GET /health", facade.HealthHandler) limitingMux.HandleFunc("POST /limit", facade.LimitHandler(limiter)) limitingSrv := &http.Server{Addr: ":8080", Handler: limitingMux, ReadHeaderTimeout: 5 * time.Second} grpcSrv := grpc.NewServer() ratelimiterv1.RegisterRateLimiterServer(grpcSrv, facade.NewGRPCServer(limiter)) tcpLis, err := net.Listen("tcp", ":8081") if err != nil { return fmt.Errorf("listen grpc: %w", err) } go func() { errCh <- limitingSrv.ListenAndServe() }() log.Printf("listening http on %s", limitingSrv.Addr) go func() { errCh <- grpcSrv.Serve(tcpLis) }() log.Printf("listening grpc on %s", tcpLis.Addr()) select { case err := <-errCh: return err case <-ctx.Done(): }sourcecraft.dev После этого создаём HTTP и gRPC сервера и ждём в select на ранее созданном NotifyContext, до момента пока не придёт сигнал на завершение процесса (os.Interrupt, syscall.SIGTERM) или ошибка. shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() grpcDone := make(chan struct{}) go func() { grpcSrv.GracefulStop(); close(grpcDone) }() select { case <-grpcDone: case <-shutdownCtx.Done(): grpcSrv.Stop() } shutdownMetrics(shutdownCtx) return limitingSrv.Shutdown(shutdownCtx)sourcecraft.dev В случае, если к нам пришла ошибка или же сигнал на завершение процесса, создаём контекст с таймаутом и просим все наши сервера выключиться. В особенности это касается gRPC сервера, он самостоятельно поддерживает GracefulStop, но если он не успеет сделать это за 10 секунд, мы выключим его принудительно. Используем ранее созданные бэкенды Помните, что мы создали Redis и InMemory бэкенды для наших алгоритмов, теперь можно посмотреть, как мы пользуемся ими, рассмотрим пример с Rolling Window: type RollingWindowStore interface { CheckAndIncrement(ctx context.Context, key string, windowSize, maxRequests, now int64) (bool, int64, error) } type RollingWindowLimiter struct { WindowSize int64 MaxRequests int64 Store RollingWindowStore Now func() int64 } func (rwl *RollingWindowLimiter) Configure() { if rwl.Now == nil { rwl.Now = func() int64 { return time.Now().Unix() } } } func ValidateRollingWindowConfig(cfg config.AlgoSettings) error { if *cfg.MaxRequests <= 0 { return fmt.Errorf("the max requests param should be strictly greater then 0") } return CheckRequiredFields(cfg, []string{"window_size", "max_requests"}) } type InMemoryRollingWindowStore struct { Backend cache.InMemoryStoreInterface } type RedisRollingWindowStore struct { Backend cache.RedisStoreInterface } func NewRollingWindowStore(store string, cacheInstance cache.InMemoryStoreInterface, redisClient *redis.Client, redisDefaultKeyTTL int64) (RollingWindowStore, error) { switch store { case "in_memory": return &InMemoryRollingWindowStore{Backend: cacheInstance}, nil case "redis": return &RedisRollingWindowStore{Backend: cache.NewRedisStore(redisClient, redisDefaultKeyTTL)}, nil default: return nil, fmt.Errorf("unsupported store: %s", store) } }sourcecraft.dev Ещё один интерфейс(RollingWindowStore), нужен для генерализации работы с бэкендом, таким образом, сам алгоритм будет просто вызывать функцию CheckAndIncrement, передавая параметры, а бэкенд будет выполнять её в атомарном варианте. func (rwl *RollingWindowLimiter) Configure() { if rwl.Now == nil { rwl.Now = func() int64 { return time.Now().Unix() } } }sourcecraft.dev Данная функция нужна для тестов, так как в тестах придётся мокать функцию получения текущего времени. type RollingWindowState struct { Timestamps []int64 } func (rws *RollingWindowState) check( windowSize, maxRequests, now int64, ) (bool, int64) { windowStart := now - windowSize i := 0 for ; i < len(rws.Timestamps); i++ { if rws.Timestamps[i] > windowStart { break } } if i > 0 { copy(rws.Timestamps, rws.Timestamps[i:]) rws.Timestamps = rws.Timestamps[:len(rws.Timestamps)-i] } if int64(len(rws.Timestamps)) < maxRequests { rws.Timestamps = append(rws.Timestamps, now) return true, 0 } retryAfter := rws.Timestamps[0] - windowStart return false, retryAfter }sourcecraft.dev Сам алгоритм устроен несложно, мы просто отслеживаем, какие временные метки уже вышли за временное окно, и если оставшихся меток меньше, чем лимит запросов, клиент может выполнить очередной запрос. Иначе нужно ответить, через какое количество секунд освободится хотя бы одно место. Далее подвязываем данную функцию к бэкенду: func (store *InMemoryRollingWindowStore) CheckAndIncrement( _ context.Context, key string, windowSize, maxRequests, now int64, ) (allowed bool, retryAfter int64, err error) { err = store.Backend.Update( key, func() any { return &RollingWindowState{Timestamps: make([]int64, 0, 8)} }, func(v any) error { st, ok := v.(*RollingWindowState) if !ok { return fmt.Errorf("unexpected state type %T for key %q", v, key) } allowed, retryAfter = st.check(windowSize, maxRequests, now) return nil }) return allowed, retryAfter, err }sourcecraft.dev То же самое в Redis: var rollingWindowScript = redis.NewScript(` local requestId = ARGV[1] local windowSize = tonumber(ARGV[2]) local maxRequests = tonumber(ARGV[3]) local now = tonumber(ARGV[4]) local windowStart = now - windowSize redis.call("ZREMRANGEBYSCORE", KEYS[1], "-inf", windowStart) local count = redis.call("ZCARD", KEYS[1]) if count < maxRequests then redis.call("ZADD", KEYS[1], now, requestId) redis.call("EXPIRE", KEYS[1], windowSize) return {1, 0} else local oldest = redis.call("ZRANGE", KEYS[1], 0, 0, "WITHSCORES") local retryAfter = 0 if oldest and oldest[2] then retryAfter = tonumber(oldest[2]) - windowStart end return {0, retryAfter} end `) func (s *RedisRollingWindowStore) CheckAndIncrement(ctx context.Context, key string, windowSize, maxRequests, now int64) (bool, int64, error) { requestId := uuid.New().String() return s.Backend.RunLuaScript(ctx, rollingWindowScript, []string{key}, requestId, windowSize, maxRequests, now) }sourcecraft.dev В Redis реализации стоит обратить внимание на то, что мы генерируемrequestId, всё потому, что мы храним наши элементы в множестве, а в множестве могут храниться только уникальные элементы. Но как тогда быть в ситуации, когда 2 и более запросов прилетают в одну секунду? Для этого и создан requestId, он добавляет уникальность каждому запросу, поэтому если в одну секунду прилетит 1000 запросов, они все посчитаются как разные запросы. А что с метриками? Как я ранее говорил, rate-limiter поддерживает подсчёт метрик. Начнём с функции instrument, о которой говорилось ранее: package metrics import ( "context" "qwrttqr-rate-limiter/core/server/algos" "time" ) type instrumented struct { next algos.Limiter rec Recorder } func Instrument(next algos.Limiter, rec Recorder) algos.Limiter { return &instrumented{next: next, rec: rec} } func (i *instrumented) Allow(ctx context.Context, req algos.Request) (algos.Decision, error) { start := time.Now() dec, err := i.next.Allow(ctx, req) elapsed := time.Since(start) switch { case err != nil: i.rec.ObserveDecision(Errored, elapsed) case dec.Allowed: i.rec.ObserveDecision(Allowed, elapsed) default: i.rec.ObserveDecision(Denied, elapsed) } return dec, err }sourcecraft.dev Данная функция просто возвращает структуру, которая обладает методом Allow, таким образом мы подменяем наш оригинальный limiter обёрткой, которая поддерживает подсчёт метрик. Recorder реализован так: package metrics import "time" type Result uint8 const ( Allowed Result = iota Denied Errored NumResults ) type Recorder interface { ObserveDecision(r Result, elapsed time.Duration) }sourcecraft.dev И конкретная реализация данного интерфейса под Prometheus выглядит так: package metrics import ( "time" "github.com/prometheus/client_golang/prometheus" ) type PromRecorder struct { decisions [NumResults]prometheus.Counter durationHist [NumResults]prometheus.Observer } func NewPromRecorder(reg prometheus.Registerer) *PromRecorder { counterVec := prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "ratelimiter", Name: "decisions_total", Help: "Each decision count by its type.", }, []string{"result"}) reqDurHistVec := prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: "ratelimiter", Name: "decision_duration_seconds", Help: "Execution time of one decision in seconds.", Buckets: []float64{0.00005, 0.0001, 0.0005, 0.001, 0.1, 0.5, 1, 2}, }, []string{"result"}) reg.MustRegister(counterVec) reg.MustRegister(reqDurHistVec) p := &PromRecorder{} p.decisions[Allowed] = counterVec.WithLabelValues("allowed") p.decisions[Denied] = counterVec.WithLabelValues("denied") p.decisions[Errored] = counterVec.WithLabelValues("errored") p.durationHist[Allowed] = reqDurHistVec.WithLabelValues("allowed") p.durationHist[Denied] = reqDurHistVec.WithLabelValues("denied") p.durationHist[Errored] = reqDurHistVec.WithLabelValues("errored") return p } func (p *PromRecorder) ObserveDecision(r Result, dur time.Duration) { p.decisions[r].Inc() p.durationHist[r].Observe(dur.Seconds()) }sourcecraft.dev Здесь мы создаём счётчики и гистограммы под каждый результат rate-limiter'а и реализуем ObserveDecision, который их обновляет. gRPC и REST gRPC Пишем .proto файл: syntax = "proto3"; package ratelimiter.v1; option go_package = "qwrttqr-rate-limiter/gen/ratelimiter/v1;ratelimiterv1"; import "google/protobuf/duration.proto"; service RateLimiter { rpc Allow(AllowRequest) returns (AllowResponse); } message AllowRequest { string key = 1; double tokens = 2; } message AllowResponse { bool allowed = 1; google.protobuf.Duration retryAfter = 2; }sourcecraft.dev Вызываем protoc и остаётся просто реализовать полученные интерфейсы: protoc -I proto --go_out=. --go_opt=module=qwrttqr-rate-limiter --go-grpc_out=. --go-grpc_opt=module=qwrttqr-rate-limiter ratelimiter/v1/ratelimiter.protosourcecraft.dev package facade import ( "context" "qwrttqr-rate-limiter/core/server/algos" ratelimiterv1 "qwrttqr-rate-limiter/gen/ratelimiter/v1" "time" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/durationpb" ) type GRPCServer struct { ratelimiterv1.UnimplementedRateLimiterServer l algos.Limiter } func NewGRPCServer(l algos.Limiter) *GRPCServer { return &GRPCServer{l: l} } func (s *GRPCServer) Allow(ctx context.Context, in *ratelimiterv1.AllowRequest) (*ratelimiterv1.AllowResponse, error) { tokens := in.GetTokens() if tokens == 0.0 { tokens = 1.0 } req, err := NewRequest(in.GetKey(), tokens) if err != nil { return nil, status.Error(codes.InvalidArgument, err.Error()) } d, err := s.l.Allow(ctx, req) if err != nil { return nil, status.Error(codes.Unavailable, "rate limiter error") } return &ratelimiterv1.AllowResponse{ Allowed: d.Allowed, RetryAfter: durationpb.New(time.Duration(d.RetryAfter) * time.Second), }, nil }sourcecraft.dev REST Для REST всё совсем просто: package facade import ( "net/http" "qwrttqr-rate-limiter/core/server/algos" "strconv" ) func HealthHandler(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) w.Write([]byte("ok")) } func LimitHandler(l algos.Limiter) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { key := r.Header.Get("X-Client-Key") tokens := 1.0 if s := r.Header.Get("X-Required-Tokens"); s != "" { n, err := strconv.ParseFloat(s, 64) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } tokens = n } req, err := NewRequest(key, tokens) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } d, err := l.Allow(r.Context(), req) switch { case err != nil: http.Error(w, "rate limiter error", http.StatusInternalServerError) // store failure case !d.Allowed: w.Header().Set("Retry-After", strconv.FormatInt(d.RetryAfter, 10)) http.Error(w, "rate limit exceeded", http.StatusTooManyRequests) } } }sourcecraft.dev Middleware Для реализации поддержки middleware можно использовать такую схему: Приложение создаёт себе бэкенд, используя Redis или оставаясь на in-memory кэше Используя созданный бэкенд создаём себе rate-limiter, с нужным алгоритмом package middleware import ( "qwrttqr-rate-limiter/core/config" "qwrttqr-rate-limiter/core/server/algos" "qwrttqr-rate-limiter/core/server/cache" "time" "github.com/redis/go-redis/v9" ) var realClock = func() int64 { return time.Now().Unix() } type Backend interface { TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error) FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error) RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error) } type memoryBackend struct { mem cache.InMemoryStoreInterface } func NewInMemoryBackend(cfg config.InMemoryBackendSettings) Backend { return &memoryBackend{mem: cache.NewCache( cfg.DefaultTtl, time.Duration(cfg.EvictionTime)*time.Second, cfg.ShardsCount), } } type redisBackend struct { redis cache.RedisStoreInterface } func NewRedisBackend(client *redis.Client, redisDefaultKeyTTL int64) Backend { return &redisBackend{redis: cache.NewRedisStore(client, redisDefaultKeyTTL)} } func newTokenBucket(cfg config.AlgoSettings, store algos.TokenBucketStore) (algos.Limiter, error) { if err := algos.ValidateTokenBucketConfig(cfg); err != nil { return nil, err } return &algos.TokenBucketLimiter{ Capacity: *cfg.Capacity, Rate: *cfg.Rate, Store: store, Now: realClock, }, nil } func newFixedWindow(cfg config.AlgoSettings, store algos.FixedWindowStore) (algos.Limiter, error) { if err := algos.ValidateFixedWindowConfig(cfg); err != nil { return nil, err } return &algos.FixedWindowLimiter{ WindowSize: *cfg.WindowSize, MaxRequests: *cfg.MaxRequests, Store: store, Now: realClock, }, nil } func newRollingWindow(cfg config.AlgoSettings, store algos.RollingWindowStore) (algos.Limiter, error) { if err := algos.ValidateRollingWindowConfig(cfg); err != nil { return nil, err } return &algos.RollingWindowLimiter{ WindowSize: *cfg.WindowSize, MaxRequests: *cfg.MaxRequests, Store: store, Now: realClock, }, nil } func (b *memoryBackend) TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error) { return newTokenBucket(cfg, &algos.InMemoryTokenBucketStore{Backend: b.mem}) } func (b *memoryBackend) FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error) { return newFixedWindow(cfg, &algos.InMemoryFixedWindowStore{Backend: b.mem}) } func (b *memoryBackend) RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error) { return newRollingWindow(cfg, &algos.InMemoryRollingWindowStore{Backend: b.mem}) } func (b *redisBackend) TokenBucket(cfg config.AlgoSettings) (algos.Limiter, error) { return newTokenBucket(cfg, &algos.RedisTokenBucketStore{Backend: b.redis}) } func (b *redisBackend) FixedWindow(cfg config.AlgoSettings) (algos.Limiter, error) { return newFixedWindow(cfg, &algos.RedisFixedWindowStore{Store: b.redis}) } func (b *redisBackend) RollingWindow(cfg config.AlgoSettings) (algos.Limiter, error) { return newRollingWindow(cfg, &algos.RedisRollingWindowStore{Backend: b.redis}) }sourcecraft.dev Далее остаётся написать функцию, создающую непосредственно саму middleware, а именно функцию возвращающую тип, поддерживающий serveHTTP метод: package middleware import ( "log" "net/http" "qwrttqr-rate-limiter/core/server/algos" "qwrttqr-rate-limiter/core/server/facade" "strconv" ) func NewHTTPMiddleware( algo algos.Limiter, keyDer HTTPKeyDerivationFunction, costDer HTTPCostDerivationFunction, failOpen bool, ) func(http.Handler) http.Handler { if algo == nil || keyDer == nil { panic("middleware: limiter and key function are required") } return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fail := func(stage string, err error) { log.Printf("limiter error on %s: %v", stage, err) if failOpen { next.ServeHTTP(w, r) return } http.Error(w, "service unavailable", http.StatusServiceUnavailable) } tokens := 1.0 if costDer != nil { t, err := costDer(r) if err != nil { fail("cost derivation", err) return } tokens = t } key, err := keyDer(r) if err != nil { fail("key derivation", err) return } req, err := facade.NewRequest(key, tokens) if err != nil { fail("request creation", err) return } decision, err := algo.Allow(r.Context(), req) if err != nil { fail("allow", err) return } if !decision.Allowed { w.Header().Set("Retry-After", strconv.FormatInt(decision.RetryAfter, 10)) http.Error(w, "rate limit exceeded", http.StatusTooManyRequests) return } next.ServeHTTP(w, r) }) } }sourcecraft.dev С gRPC необходимо реализовать 2 вида interceptor'а - унарный и потоковый: func NewGRPCUnaryInterceptor( algo algos.Limiter, keyDer GRPCKeyDerivationFunction, costDer GRPCCostDerivationFunction, failOpen bool, ) grpc.UnaryServerInterceptor { check := newGRPCCheck(algo, keyDer, costDer, failOpen) return func( ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler, ) (any, error) { if err := check(ctx); err != nil { return nil, err } return handler(ctx, req) } } func NewGRPCStreamInterceptor( algo algos.Limiter, keyDer GRPCKeyDerivationFunction, costDer GRPCCostDerivationFunction, failOpen bool, ) grpc.StreamServerInterceptor { check := newGRPCCheck(algo, keyDer, costDer, failOpen) return func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { if err := check(ss.Context()); err != nil { return err } return handler(srv, ss) } }sourcecraft.dev И check: func newGRPCCheck( algo algos.Limiter, keyDer GRPCKeyDerivationFunction, costDer GRPCCostDerivationFunction, failOpen bool, ) func(ctx context.Context) error { if algo == nil || keyDer == nil { panic("middleware: limiter and key function are required") } return func(ctx context.Context) error { fail := func(stage string, err error) error { log.Printf("limiter error on %s: %v", stage, err) if failOpen { return nil } return status.Error(codes.Unavailable, "service unavailable") } tokens := 1.0 if costDer != nil { t, err := costDer(ctx) if err != nil { return fail("cost derivation", err) } tokens = t } key, err := keyDer(ctx) if err != nil { return fail("key derivation", err) } req, err := facade.NewRequest(key, tokens) if err != nil { return fail("request creation", err) } decision, err := algo.Allow(ctx, req) if err != nil { return fail("allow", err) } if !decision.Allowed { md := metadata.Pairs("retry-after", strconv.FormatInt(decision.RetryAfter, 10)) _ = grpc.SetTrailer(ctx, md) return status.Error(codes.ResourceExhausted, "rate limit exceeded") } return nil } }sourcecraft.dev Как вы могли заметить, везде можно передать функции извлечение ключа и стоимости запроса. Это нужно так как в режиме middleware вы чаще всего будете писать сервисы, которые смотрят наружу, а значит будут принимать запросы от недоверенных клиентов. Итог Статья получилась достаточно объёмной, надеюсь она будет полезна ребятам, которые только начинают учить Golang) исходник
Two Data Sources, Zero Orchestrators — Building an F1 Analytics Pipeline on AWS Glue > Disclaimer: I'm not affiliated with or endorsed by AWS. This is a personal hobby project, and some of the behaviour described here was found by trial and error rather than from the docs. Spark Declarative Pipelines is new in AWS Glue 6.0 and may change, so check the official AWS Glue documentation first. In my first Spark Declarative Pipelines post I walked through the basic idea: declare the tables you want, let AWS Glue 6.0 work out the dependency order, and skip the orchestrator entirely. That demo used one CSV source flowing bronze → silver → gold. It worked, but it was a tidy example — one source, one shape, one straight line. Real pipelines are messier. So I built one properly, using Formula 1 race data, and deliberately made it harder: data arriving from a live API as deeply nested JSON, a second completely unrelated data source, and dimension tables that need to be deduplicated. This post walks through that build, and the things I only learned by breaking it. New to SDP? The short version: Spark Declarative Pipelines (SDP) is a framework in AWS Glue 6.0 where you declare tables instead of writing scripts that build them in order. You reference one table from another by name, and the engine figures out the dependency graph, the execution order, and what can run in parallel — all inside a single Glue job. My first post covers the fundamentals if you want them. First, the use case: what are we actually building? Let's park AWS for a minute and talk about the data, because the data shape is what drives every design decision later. I'm using f1api.dev, a free public Formula 1 API. You ask it for one race at a time: GET https://f1api.dev/api/2024/1/race That's season 2024, round 1 — the Bahrain Grand Prix. What comes back is a single, deeply nested JSON document describing everything about that race. Trimmed down heavily for readability: { "season": 2024, "races": { "raceName": "Bahrain Grand Prix", "date": "2024-03-02", "round": "1", "circuit": [ { "circuitId": "bahrain", "circuitName": "Bahrain International Circuit", "city": "Sakhir", "country": "Bahrain", "corners": 15 } ], "results": [ { "position": "1", "grid": "1", "points": 26, "driver": { "driverId": "max_verstappen", "name": "Max", "surname": "Verstappen", "number": 1 }, "team": { "teamId": "red_bull", "teamName": "Red Bull Racing" } }, { "position": "2", "points": 18, "driver": { "driverId": "perez", "...": "..." }, "...": "..." } ] } } Look at the shape, because this is the first real problem: - Race facts live at the top (name, date, round) - The circuit is an array nested one level down - The results are an array of 20 entries — one per driver — and inside each entry there's a nested driver object and a nested team object You cannot answer "which driver scored the most points this season?" from a folder full of documents shaped like that. You need flat tables. That flattening is the pipeline's first job. The questions I want answered The whole point of the pipeline is to produce tables that answer questions like: - Which driver scored the most points, won the most races, and had the best average finishing position? - Which team scored the most, and how many different drivers did they field? - Which circuits have been raced at, how many times, and what's the lap record there? Why three layers? This is the medallion architecture — bronze, curated (often called silver), and gold. It sounds formal, but the idea is simple: don't try to go from raw JSON to a polished dashboard table in one step. Do it in three, where each step has one job. Bronze — land it, then flatten it. First, raw_race stores the API response more or less as it arrived. One row per pipeline run. Nothing cleaned, nothing dropped. If I later discover a bug in my parsing logic, the original data is still sitting here and I can reprocess it. Then four tables flatten that nesting into something usable: | Table | What one row means | |---|---| | race_results_unnested | One driver's result in one race (20 rows per race) | | race_drivers | A driver's profile, as it appeared in that race | | race_teams | A team's profile, once for each of its drivers in that race | | race_circuits | The circuit that race was held at | !Bronze layer ER diagram: raw_race flattened into four prepared tables, with circuit_track_type shown unconnected Curated — one row per thing. Here's the catch with that bronze layer: every single race response re-sends the full profile of every driver and team. Load 24 races and you have 24 copies of Max Verstappen's profile and about 48 copies of Red Bull's team record (one for each of its two drivers, every race) — most of them identical, some slightly different as the season progresses and championship counts change. For analytics I want exactly one current row per driver. So the curated layer deduplicates: - dim_drivers — one row per driver (the most recent version of their profile) - dim_teams — one row per team - dim_circuits — one row per circuit - fact_race_results — one row per (race, driver): the actual measurements — grid position, finishing position, points scored If you've seen a star schema before, that's exactly what this is: dimensions describe the who/what/where, and the fact table records what happened, with the numbers. !Curated layer ER diagram: three dimension tables joined to the fact_race_results table Gold — the answers. Three summary tables, each one a direct answer to one of the questions above: driver_performance_summary, team_performance_summary, and circuit_summary. These are what a dashboard or an Athena query would actually read. !Gold layer ER diagram: three standalone summary tables One run = one race One design decision worth calling out because it shapes everything: each pipeline run fetches exactly one race. You tell it the season and round when you start the job: scripts/run_pipeline.sh --profile my-profile --season 2024 --round 1 --mode run --wait Run it again with --round 2 and the pipeline accumulates — bronze grows by one more race, and the curated and gold layers recompute to include it. After running rounds 1 and 2, my gold table showed Verstappen on 2 wins and 51 points, Perez on 36, Leclerc on 28 — which matches the real 2024 championship standings after two races — always a nice moment when the numbers come out right. !Athena query results: driver standings after two races, Verstappen first on 51 points The pipeline at a glance All of the above runs as one Glue job. Not one job per layer, not a Step Functions state machine tying them together — one job, with all tables registered as Apache Iceberg tables in the Glue Data Catalog. !Glue console Runs tab for f1-sdp-job showing a successful run and its input arguments.The red box shows the job's arguments: --enable-spark-declarative-pipeline is what switches SDP on, and the --conf values are the season and round I passed for this run. In SDP you write two kinds of tables, and the difference matters: - Streaming tables process only what's new since the last run. Bronze uses these, so re-running never reprocesses an old race. - Materialized views fully recompute every run. Curated and gold use these, because "the latest row per driver" and "total points per driver" genuinely need to look at everything each time. !Glue Data Catalog table list for f1_sdp_db showing 13 tables Adding a second data source (and why F1 needed one) Here's where I actually tested the "zero orchestration" claim. The F1 API tells me a circuit's name, city, country, length, and corner count. It does not tell me whether it's a street circuit (Monaco, Singapore, Las Vegas — public roads closed for the weekend) or a permanent racetrack (Silverstone, Spa, Suzuka — purpose-built facilities). That's exactly the kind of attribute you'd want to slice results by — for example, comparing how drivers perform on street circuits versus permanent ones. That data simply isn't in the API. It has to come from somewhere else — which makes it a perfect second data source, and a genuinely different kind of source: | | raw_race | circuit_track_type | |---|---|---| | Where it comes from | Live external API | Static CSV files in S3 | | How it arrives | An HTTP call during the job run | Uploaded by a deploy script | | Depends on | Nothing — it's a root | Nothing — also a root | Two completely independent branches. They never touch in bronze. They meet for the first time in the curated layer, in exactly one place: CREATE MATERIALIZED VIEW dim_circuits AS SELECT rc.circuit_Id, rc.name, rc.city, rc.country, ctt.track_type, rc.length, rc.corners, ... FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY circuit_Id ORDER BY race_Date DESC, load_date_time DESC) AS _rn FROM race_circuits ) rc LEFT JOIN ( SELECT *, ROW_NUMBER() OVER (PARTITION BY circuit_Id ORDER BY load_date_time DESC) AS _rn FROM circuit_track_type ) ctt ON rc.circuit_Id = ctt.circuit_Id AND ctt._rn = 1 WHERE rc._rn = 1; Ignore the ROW_NUMBER() noise for a second (that's the deduplication I mentioned — "keep the newest row per circuit"). The important line is LEFT JOIN ... circuit_track_type. That join is the entire integration. I didn't register the new source anywhere. No config file entry, no trigger, no second job, no DAG definition. SDP reads this SQL file, sees a reference to a table called circuit_track_type, and from that alone works out that it's an upstream dependency of dim_circuits. The two branches stay independent until this join — and per the AWS docs, SDP runs independent branches in parallel. That's the claim from my first article, and this is the first time I'd properly stress-tested it. It held. !Athena query results: bahrain is a permanent circuit and jeddah a street circuit Four lessons the tidy demo didn't teach me A one-source demo hides most of the real friction. These four cost me the most time. Here they are at a glance, then in detail. | # | The lesson | What I saw | What worked | |---|---|---|---| | 1 | No MERGE or upsert | No built-in way to keep a row's history | Keep only the latest row per key (SCD Type 1) | | 2 | Streaming tables need a streaming source | An INVALID_FLOW_QUERY_TYPE error on the API data | Stage the API response in S3, then read it as a stream | | 3 | Fresh deployments can fail | PATH_NOT_FOUND before any data had landed | Write a tiny .keep marker file into each watched folder | | 4 | Shared code isn't importable | ModuleNotFoundError | One shared bootstrap file, listed first in the manifest | 1. There's no MERGE or upsert, so there's no change history The problem. Databricks' version of declarative pipelines has a built-in feature for tracking how dimension rows change over time (create_auto_cdc_flow). AWS Glue 6.0 SDP has no equivalent. Why. SDP gives you only two kinds of persisted table to write to: append-only streaming tables and full-recompute materialized views. There's no "update this row" operation. What worked. I settled for SCD Type 1: keep only the current version of each row, with no history. (SCD Type 2 would keep every past version.) Every curated table works out "latest row per key" with a ROW_NUMBER() window function over the whole bronze history — number each driver's rows newest-first, then keep row 1. It works well. But if you genuinely need history ("what was this team's championship count last month?"), know that going in: there's no built-in escape hatch yet. 2. A streaming table can't be fed from a plain API call The problem. My first instinct was the obvious one: fetch the JSON from the API, load it into a DataFrame in memory, and return it from the pipeline. Glue rejected it: AnalysisException: [INVALID_FLOW_QUERY_TYPE.BATCH_RELATION_FOR_STREAMING_TABLE] Why. A streaming table has to be fed by a streaming source: something Spark can read incrementally while remembering how far it got. A DataFrame built in memory is a one-off batch, so it's refused. What worked. Stage the data in S3 first: 1. Call the API and get the JSON. 2. Write it to a uniquely named file in an S3 folder, using plain boto3 (not Spark). 3. Read that folder with spark.readStream. Now it's a genuine streaming source, and Spark's built-in file checkpointing means a re-run never reprocesses yesterday's file. It's an extra moving part that exists purely because a REST API has no natural streaming shape. 3. A brand-new deployment fails before it does anything The problem. On a fresh deploy, the pipeline failed straight away: AnalysisException: [PATH_NOT_FOUND] Path does not exist: s3://.../reference-data/circuit_track_type/ Why. Both S3 folders the pipeline watches (the API staging folder and the reference-data folder) are read with spark.readStream(...).load(path). Before any data has landed, those folders don't exist, and Spark doesn't shrug and read zero rows. It fails the job. What worked. Make sure each folder always exists by writing a tiny empty marker file, .keep, into it. My CDK stack does this for the reference-data folder on every deploy, and the pipeline does it itself for the API staging folder. It's cheap insurance against a failure that only appears on day one of a new environment, which is exactly when you're least expecting it. 4. Shared Python code isn't importable by default The problem. I keep schema definitions and API-fetch helpers in a shared Python package next to the transformation files. Importing it failed: ModuleNotFoundError: No module named 'f1_pipeline_lib' Why. SDP loads each transformation file directly by its file path, and that doesn't add anything to Python's import path (I confirmed this by reading SDP's source). I then tried the documented way to ship extra Python files to a Glue job, the --extra-py-files parameter. It failed with the same error, most likely because SDP runs as a separate process that the setting doesn't reach. I confirmed both failures by running real Glue jobs and reading the tracebacks. What worked. One tiny shared bootstrap file that fixes the import path, listed first in the pipeline manifest so it always runs before anything else. One file, one place, instead of the same few lines copy-pasted into every transformation file. Benefits of using SDP in AWS Glue 6.0 - The zero-orchestration claim genuinely holds, even with two independent sources and a deduplicating curated layer. One job, no Step Functions, no Airflow. - Connecting a new source takes one line. The source still needs its own table definition, but wiring it into the pipeline is just a SQL JOIN or a .readStream.table("x") call — there's nothing to register anywhere else. - Independent branches run in parallel automatically, according to the AWS docs — no fan-out/fan-in code to write. - SQL and Python mix freely in the same pipeline — my bronze layer is Python, curated and gold are plain SQL files. - Streaming tables remember what they've already processed. Each one keeps a checkpoint in S3 — a record of which input it has already consumed — so every run picks up only what's new. I checked this directly on the reference-data table: after adding a second CSV file, the checkpoint's log showed the next run read just that new file, not the original again. The same mechanism means running round 2 adds that race alongside round 1 instead of starting over. It isn't free in the cost sense — you still pay for the compute that runs — but you don't write or maintain any "last processed" bookmark yourself, which is easy to get subtly wrong in a hand-rolled pipeline. (Materialized views don't get this; they recompute from scratch every run, which is the trade-off listed under limitations below.) - Tables register themselves in the Glue Data Catalog — no manual CREATE TABLE anywhere. - VALIDATE mode is a genuinely useful dry run that catches dependency and compile errors before you spend money on a full run. Limitations of using SDP in AWS Glue 6.0 Three of these — materialized views always recomputing, no data quality expectations, and the .withColumn() gotcha — are also listed under Considerations and limitations in the official AWS docs. - No CDC/MERGE/upsert. No SCD Type 2 out of the box, as described above. - No data quality expectations. Inline data quality annotations are not supported in the current SDP framework. Some other declarative frameworks let you attach rules like "this column must not be null" directly to a table definition; SDP in AWS Glue 6.0 has no equivalent yet, so any validation is hand-written. My pipeline doesn't have a quality layer yet, so a malformed API response simply fails the run. - Materialized views always fully recompute. Only streaming tables are incremental — worth budgeting for as your gold aggregations grow. - Materialized views can read as empty while a job is running. Every run clears each materialized view first (a TRUNCATE TABLE), and the new rows only land when that view's own step finishes. So an Athena query during a run returns zero rows — not an error, and not the previous data — until the refresh completes. I checked this on a real run of this pipeline by polling Athena every few seconds: each of my four materialized views read 0 rows for roughly 30 seconds of a two-minute run, with the gold tables coming back last, while a streaming table stayed readable the whole time. I couldn't find this in the AWS Glue or Apache Spark documentation; I found the cause in Spark's open-source SDP code. In Spark 4.1.1, the version Glue 6.0 runs, it truncates every non-streaming table at the start of an update. Plan for it: don't point a dashboard or a downstream job at these tables while a run is in progress. - External API sources need a staging workaround to satisfy the streaming-relation requirement. - Python packaging has sharp edges that I couldn't find documented. The import-path issue, and the fact that --extra-py-files doesn't reach the SDP process, both cost me real debugging time. - A .withColumn() gotcha. If a downstream table reads another pipeline table and adds a derived column with .withColumn(...), SDP might fail to detect the dependency on the second and later runs, so the downstream table reads stale data from the previous run (a one-run lag). The documented fix is to build derived columns inside a single .select(...) instead, and to avoid operations that force plan resolution, such as .schema or .collect, inside query functions. I followed this from the start — every derived column in my bronze layer is written in a .select(...) — so I never saw the stale-data problem itself. I did hit a similar sharp edge in the same place: adding .distinct() to a streaming read made the pipeline fail with a misleading "table not found" error, and removing it fixed it. - VALIDATE still runs your code; it just doesn't write the results. A live-API source will happily call the real API during a "dry run" unless you guard against it. What I took away from this My first article proved the concept worked. This one was about whether it survives something less convenient — a second unrelated source, nested data that needs real flattening, and state that has to stay correct across many runs. It does, and the "add a source, write one join" story held up exactly as advertised. That still impresses me. What surprised me is where the effort actually went. Almost none of it was in the declarative pipeline logic — that part really is as clean as promised. It all went into the handful of places where SDP's abstractions meet the realities of Glue and Spark underneath: the streaming-source rule, the missing-folder failure, and the import path. And none of those were things I could have read my way to. Every one came from running a real job, reading the actual CloudWatch traceback, and occasionally reading the framework's own source code to find out what it was really doing. If you're evaluating SDP for anything beyond a demo, that's my main advice: budget time for that kind of hands-on verification. The gap between "the docs say" and "a real Glue 6.0 job does" is still a real gap today. Just know that the core promise — one job, no orchestrator, dependencies figured out for you — does deliver. The code for this project is on GitHub. References - "One Glue Job, Zero Orchestrators — My First Spark Declarative Pipeline on AWS Glue", DEV Community - "Build declarative ETL pipelines with AWS Glue 6.0", AWS Big Data Blog - AWS Glue Spark Declarative Pipelines documentation - f1api.dev — the free Formula 1 API used in this project --- About the author I'm an AWS Community Builder in the Data category. I build small, real projects and write up what the documentation didn't cover. Questions, or something broke differently for you? Connect with me on LinkedIn or drop a comment below.
AUTO_CDC: Turning an event log into current state with Spark Declarative Pipelines The question an event log can't answer cheaply The dataset I use for most of my pipeline demos is a simulated ghost kitchen (a delivery-only restaurant operation), and like most operational data it arrives as an append-only log. Every order produces a lifecycle of events, each stamped with a time and a per-order sequence number, and a day of it is about a million events. The generator deliberately injects a few percent of late and duplicated events, because real streams do that. Logs like this are excellent at answering what happened and awkward at answering what operations people actually ask: which orders are live right now, and where is each one? The usual answer is to recompute it. Group the whole history by order, take the latest event for each, drop the cancelled ones, and repeat on every refresh, paying for the entire history every time and getting it subtly wrong whenever "latest" means latest to arrive rather than latest to happen. The better answer depends on noticing something that sounds obvious once it's said: an append-only lifecycle log is already a change data capture feed. Each event is a change to the current state of one entity (the order), keyed by order id, sequenced by when it happened, and a cancellation is simply a delete. Here is an illustrative order (simplified; the real sequence is described below), first as the lifecycle log records it, then as the change feed it implies: | seq | event_type | as a change to orders_live | |---|---|---| | 1 | order_created | UPSERT order 1042, status = order_created | | 2 | kitchen_started | UPSERT order 1042, status = kitchen_started | | 3 | order_ready | UPSERT order 1042, status = order_ready | | 4 | driver_picked_up | UPSERT order 1042, status = driver_picked_up | | 5 | driver_ping | UPSERT order 1042, status = driver_ping | | 6 | delivered | UPSERT order 1042, status = delivered | An order that ends in order_cancelled produces a DELETE as its last change instead, and disappears from the table. Once the log is seen this way, "current status of every order" stops being an aggregation problem and becomes the problem of applying a change feed to a table, which is exactly what AUTO CDC was built to do. My argument is that AUTO CDC is worth adopting less because it saves code (though it does, a lot) than because it moves the hard parts of change application into the engine, and leaves you with a small, explicit contract: a sequence column that is unique and never null, a target that genuinely supports row-level operations, and deletes expressed as events. Most of what goes wrong with hand-rolled CDC is the engine's side of that contract done badly; most of what can still go wrong with AUTO CDC is your side of it. What AUTO CDC is, and where it stands AUTO CDC began as APPLY CHANGES in Databricks' Delta Live Tables and was contributed to Apache Spark under its own SPIP (SPARK-56249), as part of Spark Declarative Pipelines. In Python it's create_auto_cdc_flow. | Release | AUTO CDC status | |---|---| | Spark 4.2.0 (July 2026) | Shipped. SCD Type 1 via the Python / Spark Connect API | | Spark 4.3.0 (release candidate) | Adds SCD Type 2 (history with validity ranges) | | main branch (5.0.0-SNAPSHOT) | Ongoing work under SPARK-56249 | Everything I tested was SCD Type 1 (the target holds only the current state per key, and updates overwrite in place), on a build of the main branch from late May 2026, before 4.2 was released. Applying a change feed correctly means doing several things at once: - matching each change to its row by key - keeping the version with the highest sequence per key, not the last to arrive - honouring deletes, but only when the delete is the latest thing that happened to that key - staying idempotent when a streaming source redelivers after a failure The hand-rolled version is usually a foreachBatch with windowed deduplication feeding a MERGE, and it tends to get the cross-batch cases wrong without throwing errors. A late update that lands in the micro-batch after a newer one quietly overwrites it, and nobody notices until the numbers disagree with something else. !diagram 1 The code The projection from lifecycle log to change feed is ordinary Spark. This is the core of what my test harness does: every event becomes an UPSERT of the order's current status, a cancellation becomes a DELETE, and the sequence is made unique per order. feed = events.select( F.col("order_id"), F.col("event_type").alias("status"), F.col("location_id").cast("int").alias("location_id"), F.col("ts_seconds").alias("event_ts"), F.when(F.col("event_type") == "order_cancelled", F.lit("DELETE")) .otherwise(F.lit("UPSERT")).alias("op"), # unique per-key ordering: event time, broken by per-order sequence (F.col("ts_seconds") * F.lit(100) + F.col("sequence")).alias("seq"), ) Then the pipeline itself is three declarations: a streaming source, a target streaming table, and the flow. from pyspark import pipelines as dp @dp.table(name="order_cdc_source") def order_cdc_source(): return spark.readStream.schema(ORDER_SCHEMA).json("file:///data/order_cdc/*.json") dp.create_streaming_table("ice.gold.orders_live") dp.create_auto_cdc_flow( target="ice.gold.orders_live", source="order_cdc_source", keys=["order_id"], # how a change finds its row sequence_by="seq", # highest sequence per key wins, whatever the arrival order apply_as_deletes="op = 'DELETE'", # which events delete the row except_column_list=["op", "seq"], # bookkeeping columns to leave out of the target stored_as_scd_type=1, ) The engine handles what the hand-rolled version gets wrong: deduplication across micro-batches by sequence_by, the generated upsert and delete, exactly-once checkpointing, and the auxiliary state table that remembers the latest sequence per key across runs. That table is what lets a late event arriving in a later batch resolve correctly. | | Recompute from the log | AUTO CDC | |---|---|---| | Cost per refresh | re-aggregates the entire history | processes only new events | | Out-of-order events | "latest" by arrival is wrong | resolved by sequence_by | | Cancellations | a row every query must remember to filter | the row is deleted | | Code | bespoke foreachBatch + dedup + MERGE | three declarations | Running it on messy data I ran the order-status projection on a sample of the ghost-kitchen stream: a little under five thousand orders and roughly 125,000 of their lifecycle events, including the injected late and duplicate ones. AUTO CDC collapsed them into one row per live order: | gold.orders_live | rows | |---|---| | delivered | 4,601 | | driver_ping (mid-delivery) | 84 | | order_ready | 3 | | order_created | 1 | | kitchen_started | 1 | | driver_picked_up | 1 | | total live orders | 4,691 | | cancelled orders removed by apply_as_deletes | 308 | To check it, I recomputed the same table from scratch the naive way and compared the two in both directions: w = Window.partitionBy("order_id").orderBy(F.desc("seq")) naive = (feed.withColumn("rn", F.row_number().over(w)).filter("rn = 1") .filter("op != 'DELETE'").select("order_id", "status")) pairs = spark.table("ice.gold.orders_live").select("order_id", "status") match = naive.subtract(pairs).count() == 0 and pairs.subtract(naive).count() == 0 # match = True They matched exactly. It's worth being precise about what that proves: it compares the two columns the table exists to answer, as sets, rather than every column byte for byte. Two details of the run also deserve honesty, because they're the kind of thing that makes a demo look cleaner than it was. The sample was drawn by sorting order ids and taking the first five thousand, and because the raw data contains events with a null order id (which Spark sorts first), the sample was really 4,999 orders. And the projection joined each event to its order's creation event, so orders with duplicated creation events fanned out into duplicate rows in the feed. AUTO CDC handled both without complaint, since duplicate copies of an event carry the same sequence value and resolve to the same row, but a pipeline that quietly processes inputs it wasn't meant to receive is a reminder to look at your inputs before trusting your outputs. The second scenario was the textbook one: keeping a dimension current as kitchens relocate, rename, and close. These are the exact changes streamed into silver.dim_locations_current, on top of an initial load of four locations at sequence 1: changes = [ {"id": 2, "city": "Los Angeles - Arts District", "op": "UPSERT", "seq": 5}, {"id": 1, "city": "San Francisco - SoMa", "op": "UPSERT", "seq": 4}, {"id": 3, "city": "Chicago", "op": "DELETE", "seq": 6}, # kitchen closed {"id": 2, "city": "Los Angeles (STALE)", "op": "UPSERT", "seq": 3}, # arrives late, older ] | id | after AUTO CDC | why | |---|---|---| | 1 | San Francisco - SoMa | rename at seq 4 applied | | 2 | Los Angeles - Arts District | rename at seq 5 kept; the stale seq 3 edit ignored | | 3 | (no row) | closure at seq 6 deleted it | | 4 | Austin | unchanged | A nightly full reload would eventually reach the same names, but it copies every unchanged row to get there and has no way at all to express that a location no longer exists. That's the more general lesson: an append-only or aggregate pipeline can only add facts, so "this was cancelled" becomes another row every downstream query has to remember to filter. In a CDC pipeline a terminal business event can be the delete itself, and the same mechanism propagates hard deletes for compliance. Edge cases, and how Type 1 resolves them I put SCD Type 1 through a battery of adversarial cases, and it behaved correctly in every one. The behaviours are worth knowing precisely, because this is where intuition tends to be wrong. These are the cases from my stress harness, with the events in arrival order; | separates micro-batches. | Case | Events for one key | Result | |---|---|---| | Upsert to a key that never existed | UPSERT@1 | row inserted | | Delete of a key that never existed | DELETE@1 | no-op | | Several updates in one batch | UPSERT@1, UPSERT@2, UPSERT@3 | row has the @3 values | | Stale delete | UPSERT@1 | UPSERT@3 | DELETE@2 | row stays, with the @3 values; the update is newer | | Stale insert after a delete | UPSERT@3 | DELETE@5 | UPSERT@2 | row stays deleted | | Delete arrives before an older insert | DELETE@5 | UPSERT@2 | row stays deleted | | Resurrection | UPSERT@1 | DELETE@2 | UPSERT@3 | row present, with the @3 values | | Volume, out of order | 500 keys, each UPSERT@1, @3, @2 shuffled across 5 batches, plus DELETE@5 for 100 of them | exactly 400 rows, all with the @3 values | | Tie (probe, not pass/fail) | UPSERT@5 "Boston", UPSERT@5 "Denver" in one batch | one of them, chosen arbitrarily | Your side of the contract The sequence. A null in the sequence_by column fails the micro-batch with AUTOCDC_MICROBATCH_VALIDATION.NULL_SEQUENCE rather than guessing, which is the right behaviour, and means the column has to be non-null by construction. Ties are worse, because they don't fail: two changes to the same key with the same sequence in one micro-batch are resolved arbitrarily, as the programming guide itself warns. A coarse timestamp is the classic way to produce ties, so sequence on something strictly increasing per key, like a log offset or commit LSN. My composite of timestamp times 100 plus the per-order sequence works only because no order in this data has more than 99 events; it's a local trick, not a general recipe. The ties that did occur in my data were all injected exact duplicates, the one case where an arbitrary choice is harmless. The target. AUTO CDC needs a connector that implements Spark's DataSource V2 row-level operation interface (SupportsRowLevelOperations), and "supports MERGE" is not the same thing: | Target | Result in my testing | |---|---| | Plain Parquet | rejected immediately: AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE | | Delta | did not work: Delta's table implementation doesn't expose SupportsRowLevelOperations, despite supporting MERGE elsewhere | | Iceberg, Hadoop catalog | worked | | Iceberg, JDBC (Postgres) catalog | worked | | Iceberg, Hive Metastore catalog | worked | | Iceberg, Gravitino REST catalog | worked | | Iceberg, Glue catalog (emulated) | worked | | Iceberg, Unity Catalog OSS REST | failed: the catalog is read-only | | Any ephemeral / in-memory catalog | unsuitable: the aux state table must persist across restarts | Two smaller surprises: the target carries a hidden __spark_autocdc_metadata column that shows up in SELECT *, and Iceberg's own changelog is batch-only in open-source Spark for now, so you can't stream one Iceberg table's changes straight into AUTO CDC as a source. A correction. I originally reported that running an AUTO CDC pipeline through the spark-pipelines CLI failed at registration with InvalidPlanInput: RELTYPE_NOT_SET, and called it a bug. It wasn't. The CLI connects to the default Spark Connect endpoint, and an old Spark 4.1 Connect server was still listening there, so my pipeline was being sent to a server that had never heard of AUTO CDC. Against a server of the right version the CLI works correctly. If you see that error, check what's actually listening on sc://localhost:15002 before anything else. What changes With a hand-rolled pipeline, the question you're left with is whether you've correctly implemented cross-batch deduplication, sequenced deletes, and idempotent merges, and the honest answer is usually that you won't know until something disagrees. With AUTO CDC that part belongs to the engine, which is tested against exactly the cases people get wrong. The question becomes whether you've given it a sequence that is unique and never null, a target that really supports row-level operations, and a delete for everything that should be allowed to disappear. That's a much better question to be left with, and the first step is often not a new data source at all; it's noticing that the event log you already have has been a change feed all along. Further reading - Test harnesses and results: lakehouse-stack/scripts/pipelines/autocdc-scd1 (Spark main branch, 5.0.0-SNAPSHOT at 06ffa273, with an Iceberg runtime ported to Spark 5.0).
From Pickle to Arrow: A Decade of Moving Data Between Spark's JVM and Python Arrow's real contribution to PySpark isn't speed at any one API but a shared memory format at every boundary between the JVM and Python; each release removed one conversion at one boundary, what each one left behind explains the next, and the conversion that remains now lives inside Spark's own engine rather than at the Python edge. That crossing is the oldest problem in PySpark, and the most persistent attempt to fix it started in February 2016, when Wes McKinney (the creator of pandas) filed a Spark ticket with a modest title: SPARK-13534, "Implement Apache Arrow serializer for Spark DataFrame for use in DataFrame.toPandas." It asked for one method to stop using pickle. It took two years to ship. And over the decade that followed, the same idea went through every other door between Spark's JVM and Python: UDFs, partition-level APIs, the network protocol of Spark Connect, custom data sources, until in Spark 4.1 a Python function could receive Spark's data as an Arrow array and never touch pandas at all, and in Spark 4.2 the Arrow paths finally became the defaults. --- The wall PySpark is not a Python reimplementation of Spark. It's a Python client for a JVM engine. When you write df.filter(col("x") > 5).groupBy("y").count(), Python builds a query plan and hands it to the JVM, which optimizes, distributes, and executes it; Python never touches the data, which is why DataFrame code runs at the same speed in PySpark and Scala. The trouble starts the moment your own code needs the data: a UDF, toPandas(), createDataFrame(pandas_df), a custom data source. Then data has to physically cross from the JVM into a Python process, and for most of Spark's history the only way across was pickle, one row at a time: !diagram 1 Every row pays four serialization steps and a function call. For a UDF like return x * 1.1, the arithmetic is nanoseconds and the machinery around it is microseconds, three orders of magnitude of overhead wrapped around almost no work. That was PySpark from 2015 to 2018, and the same crossing existed at every door between the two runtimes, on both the driver and the executors: !diagram 2 Why the fix was a memory format, not a faster pickle Arrow was conceived in late 2015 by people who had hit the same wall from different directions. McKinney had spent years fighting pandas' memory model; his 2017 essay "Apache Arrow and the 10 Things I Hate About pandas" lists the grievances, including a rule of thumb that pandas users needed "5 to 10 times as much RAM as the size of your dataset." Jacques Nadeau, building a query engine at Dremio, saw every data system inventing its own internal format and paying to translate between them at every boundary. Marcel Kornacker, who created Impala at Cloudera, saw the same copying across the Hadoop ecosystem. They met at Strata NYC in September 2015, and Arrow was announced as an Apache top-level project in February 2016, skipping the incubator entirely, which is unusual and reflected how many companies wanted the same thing. The idea was not a faster serializer. It was to agree on the bytes themselves: a language-independent, columnar, in-memory layout specified down to byte alignment, null bitmaps, and buffer order, so that two Arrow-aware systems can hand each other data without converting it. Within a process, or with shared memory, that's genuinely zero-copy; across a socket it's still a copy, but with no per-row decoding on either side, because the bytes on the wire are already the bytes in memory. Columnar mattered as much as standard. In a row layout, values from one column are interleaved with everything else in the row: row layout: [id=1, name="Alice", age=30] [id=2, name="Bob", age=25] [id=3, name="Carol", age=35] columnar: id: [1, 2, 3] name: ["Alice", "Bob", "Carol"] age: [30, 25, 35] With columns stored contiguously, a 64-byte cache line holds eight useful int64 values instead of one row's mixed fields, and SIMD instructions can process a run of values at once. The database literature had already shown what this buys: the MonetDB/X100 work (CIDR 2005) argued that pushing batches of around a thousand column values through each operator, instead of one row at a time, is an order of magnitude or two faster. Arrow's batch-oriented design was built around exactly that model. --- Boundary one: the driver (2018) The first door was the smallest. DataFrame.toPandas() collected results to the driver by pickling every row and rebuilding each one as Python objects before pandas could wrap them, and it was notoriously slow. McKinney's ticket, filed on February 28, 2016, three weeks after the first commit to the Arrow repository, asked to replace that path. It stayed open for over a year. Bryan Cutler, at IBM's Spark Technology Center, resolved it on June 23, 2017, and it reached users in Spark 2.3.0 in February 2018. Two years, for one method. !diagram 3 # Spark 2.3 name, later renamed; on by default since Spark 4.2 spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true") The reverse direction, createDataFrame(pandas_df), got the same treatment (SPARK-20791). What made this matter wasn't the one method; it was the plumbing Cutler built to do it (Arrow converters on the JVM side, pyarrow integration on the Python side), which every later boundary reused. What it removed: pickle for collecting data to the driver. What it left: everything that happens on executors. UDFs still crossed row by row. Boundary two: UDFs, in batches (2018) The second door was the one that mattered for performance, and it came from outside Databricks. The proposal, SPARK-21190 ("SPIP: Vectorized UDFs in Python"), was filed by Reynold Xin, but most of the implementation was done by Li Jin, an engineer at the quantitative investment firm Two Sigma and an Arrow committer, on top of Cutler's serializer. It shipped in the same Spark 2.3.0 release. Instead of calling your function once per row, Spark would send a batch of rows (10,000 by default, spark.sql.execution.arrow.maxRecordsPerBatch) as an Arrow RecordBatch and call your function once per batch, with a pandas Series: # Before: once per row, pickle both ways @udf(returnType="double") def add_tax(price): return price * 1.1 if price else None # After: once per batch, a pandas Series in and out @pandas_udf("double") def add_tax(price: pd.Series) -> pd.Series: return price * 1.1 Li Jin's post introducing the feature in October 2017 reported speedups from around 3x on simple arithmetic to around 100x on a cumulative-probability function, and the spread is the point: the simpler the function, the more of its cost was the crossing, and the more complex the function, the more it also benefits from running as vectorized NumPy instead of Python bytecode. My own measurements, eight years and many releases later, land in the same shape. At 50 million rows, a pandas UDF did the arithmetic workload in 0.79 seconds against 5.20 for a pickled UDF, and a normal CDF in 0.82 against 6.18; on a string transform the gap was narrower, 2.51 against 6.22, for a reason that turns out to matter later. What it removed: per-row serialization and per-row function calls. What it left: a rewrite (your function had to speak pandas), and a conversion. Every batch arrived as Arrow and was turned into pandas before your code saw it, and the result was turned back: !diagram 4 What actually crosses the socket Every boundary from here on sends the same thing over the socket: Arrow's IPC stream format, a schema message followed by record-batch messages, each framed like this: ┌─────────────────────────────────────────────────────────────┐ │ Continuation indicator: 0xFFFFFFFF (4 bytes) │ │ Metadata length (4 bytes, little-endian int32) │ │ Flatbuffer metadata (Message type, schema or batch header) │ │ Padding to 8-byte boundary │ │ Message body (data buffers, padded to 8-byte multiple) │ └─────────────────────────────────────────────────────────────┘ Inside the body, each column is a set of plain buffers: a validity bitmap with one bit per value, the values themselves as a contiguous typed array, and, for variable-length types like strings, an offsets array into one contiguous UTF-8 buffer. For four rows of (name: string, age: int32): Validity bitmap (age): [1, 1, 0, 1] → 0b1011 (bit-packed) Data buffer (age): [30, 25, 0, 35] → 16 bytes (4 × int32) Offsets (name): [0, 5, 8, 13, 18] → 5 × int32 Data buffer (name): "AliceBobCarolDavid" (contiguous UTF-8) No per-value type tags, no escaping, no framing between values. Strings are where Arrow and pandas differ most: in Arrow, a million strings are one buffer and an array of offsets, while in traditional pandas they're a million separate Python objects. --- Boundary three: letting code see Arrow (2022 to 2025) For four years, every batch API in PySpark handed you pandas, whether you wanted pandas or not. The first API that let user code see Arrow directly was mapInArrow, added by Hyukjin Kwon, a long-time PySpark maintainer at Databricks (SPARK-37227, with the Python side in SPARK-37228), in Spark 3.3.0 in June 2022. It passes whole partitions as an iterator of pyarrow.RecordBatch objects, and your function yields RecordBatches back: import pyarrow as pa import pyarrow.compute as pc def add_tax_arrow(batches): for batch in batches: taxed = pc.multiply(batch.column("price"), pa.scalar(1.1)) yield pa.RecordBatch.from_arrays([batch.column("id"), taxed], names=["id", "price_with_tax"]) df.mapInArrow(add_tax_arrow, schema="id long, price_with_tax double") Grouped operations were the obvious remaining gap, and Enrico Minack closed it with applyInArrow (SPARK-40559), which shipped in Spark 4.0.0 in May 2025. Where applyInPandas turns each group into a pandas DataFrame, applyInArrow hands you the group as a pyarrow.Table: def normalize(table: pa.Table) -> pa.Table: totals = table.column("total") mean, std = pc.mean(totals).as_py(), pc.stddev(totals).as_py() normalized = (pa.array([0.0] * len(totals)) if not std else pc.divide(pc.subtract(totals, pa.scalar(mean)), pa.scalar(std))) return pa.Table.from_arrays([table.column("order_id"), totals, normalized], names=["order_id", "total", "normalized"]) orders.groupBy("location_id").applyInArrow( normalize, schema="order_id long, total double, normalized double") The same idea reached the other direction in 4.0 too: the new Python Data Source API (SPARK-44076) lets a custom source written in Python yield pyarrow.RecordBatch objects instead of tuples (SPARK-48493), so data read by Python code can enter Spark as Arrow instead of being built up one Python tuple at a time. I haven't benchmarked the grouped or data-source paths myself, so there are no numbers for them here; the mechanism is the same as everywhere else, and the benefit should be largest where the pandas conversion is most expensive, which is strings. What it removed: the forced detour through pandas, for anyone willing to write Arrow code. What it left: that condition. Most PySpark code is plain @udf functions that nobody is going to rewrite. Boundary four: the network (2023) Spark Connect started as a proposal from Martin Grund at Databricks in June 2022 (SPARK-39375): separate the client from the server, so an application talks to Spark over gRPC instead of living inside the driver's JVM. Once there's a network between you and the engine, the format results travel in stops being an optimization and becomes the protocol. The answer was Arrow. !diagram 5 The client sends unresolved logical plans as Protocol Buffers; the server optimizes and runs them and streams the results back as Arrow record batches over the same channel. That's what makes a JVM-free Python client possible at all: the pyspark-client package is a 1.6 MB wheel of pure Python. The same columnar format that moves data between executors and Python workers now moves data between a remote cluster and a laptop. Boundary five: the UDFs nobody rewrote (2023 to 2026) Every boundary so far helped people who changed their code. Arrow-optimized Python UDFs (SPARK-40307, Spark 3.5.0, September 2023) took the opposite approach: keep your plain @udf exactly as written, and swap the transport underneath it. Batches cross as Arrow; Python unpacks each batch and still calls your function once per value. !diagram 6 Xinrong Meng's 2023 Databricks post measured about 1.6x over pickle for a single UDF and 1.9x for three chained ones on a 32 GB dataset, and my benchmark agrees in spirit: useArrow=True took the arithmetic workload from 5.20 to 2.23 seconds and the CDF from 6.18 to 2.94. Real, and cheap to get, but nothing like the pandas UDF numbers, because the per-row function call is still there. That call is my benchmark's ceiling. Turning this on by default took two tries. It was merged for Spark 4.0 (SPARK-48516) and reverted in March 2025 before the release; it finally shipped as the default in Spark 4.2, along with Arrow for toPandas() and for Python UDTFs, and a minimum PyArrow of 18.0. On 4.2, then, every existing @udf gets Arrow transport without anyone touching it. It's also where I made the mistake that best illustrates how confusing this boundary is. My benchmark's "Arrow" variant of the plain UDF flipped spark.sql.execution.arrow.pyspark.enabled, which governs toPandas() and createDataFrame() and has nothing to do with UDFs. The setting that actually moves a UDF onto Arrow is spark.sql.execution.pythonUDF.arrow.enabled, or useArrow=True on the decorator. So my "Arrow" row was the pickled UDF measured twice, and its near-identical result was an accidental A/A test. Three settings are named for Arrow, and they control three different boundaries. What it removed: pickle, for everyone, with no code changes. What it left: the function call per row, which is the part no transport format can fix. Boundary six: Arrow as the execution format (2025) The last step turned Arrow from the way data travels into the way your code computes. Ruifeng Zheng's SPARK-53014, "Make Arrow UDF public," shipped in Spark 4.1.0 on December 16, 2025, almost ten years after McKinney's ticket. An Arrow UDF receives pyarrow.Array values and returns one, and it's meant to be written with pyarrow.compute, whose functions are C++ kernels operating directly on Arrow buffers: import pyarrow as pa import pyarrow.compute as pc from pyspark.sql.functions import arrow_udf @arrow_udf("double") def add_tax(price: pa.Array) -> pa.Array: return pc.multiply(price, pa.scalar(1.1)) @arrow_udf("string") def clean_email(email: pa.Array) -> pa.Array: return pc.utf8_trim(pc.utf8_lower(email)) The four UDF types differ only in where the data changes format: | UDF type | Data path | Format conversions | |---|---|---| | Pickle @udf | JVM → pickle → Python value → pickle → JVM | pickle both ways, per row | | Arrow-optimized @udf | JVM → Arrow → Python value → Arrow → JVM | Arrow ↔ Python objects, per row | | @pandas_udf | JVM → Arrow → pandas → Arrow → JVM | Arrow ↔ pandas, per batch | | @arrow_udf | JVM → Arrow → your function → Arrow → JVM | none on the Python side | In my benchmark, @arrow_udf was the fastest Python option everywhere, but the margin over pandas UDFs depended almost entirely on the data type. On arithmetic and the CDF it was only 5 to 9% faster; on the string transform it was 1.6x faster at both 50M and 100M rows (1.55 seconds against 2.51 at 50M). The split comes down to strings. A numeric pandas Series is a NumPy array underneath, so converting from Arrow is cheap; a string column in traditional pandas is an array of Python objects, so converting means building millions of them. pyarrow.compute skips all of that and works on the one contiguous buffer. What it removed: the last conversion on the Python side. What it left: Spark's side. Taken together, the choice today comes down to what your function can receive: !diagram 7 The pipe itself Everything above changes what goes through the socket. Hyukjin Kwon's SPARK-51688, shipped in Spark 4.1.0, changes the socket: instead of TCP over loopback, which runs every byte through the full TCP/IP stack even though it never leaves the machine, the JVM and Python workers can talk over a Unix domain socket, which goes straight through a kernel buffer. It's opt-in (spark.python.unix.domain.socket.enabled, whose default comes from the PYSPARK_UDS_MODE environment variable) in 4.1 and 4.2, and that's unchanged on the 4.3 branch, which instead adds another transport option: pipelined, asynchronous transfer of UDF batches between the JVM and Python (spark.python.udf.pipelined.enabled, off by default). !diagram 8 Spark 1.x: pickle format + TCP loopback slow format over a slow pipe Spark 2.3: Arrow format + TCP loopback fast format over a slow pipe Spark 4.1: Arrow format + Unix domain socket fast format over a faster pipe (opt-in) --- One boundary at a time !diagram 9 As of Spark 4.2, Arrow is the format at every place data crosses between the JVM and Python: | Boundary | Arrow's role | Default | |---|---|---| | toPandas() / createDataFrame() | replaces pickle for driver data collection | yes, since 4.2 | | @udf (plain Python UDFs) | replaces pickle transport | yes, since 4.2 | | @pandas_udf, mapInPandas, applyInPandas | batch transport, converted to pandas | always | | @arrow_udf, mapInArrow, applyInArrow | transport and execution format | always | | Spark Connect results | the result protocol | always | | Python Data Source API | source data format | when the source yields RecordBatches | The last boundary What's left is on Spark's side of the wall. Inside the JVM, Spark still executes on its own row format (UnsafeRow, from the Tungsten project), and Arrow exists at the edges, so every crossing still begins and ends with a conversion between Spark's rows and Arrow's columns. Engines like Velox and DataFusion show that a query engine can execute on Arrow-format data directly; Spark doesn't, and closing that gap is a much bigger change than any of the boundaries above. !diagram 10 The 4.3 branch, which hasn't been released yet (its first release candidate failed its vote on September 6, and a second, put to a vote on October 13, drew -1 votes over critical fixes that still need backporting, so there's no date), has two small signs of movement at exactly this edge. One is an Arrow format for Spark's in-memory cache (ArrowCachedBatchSerializer, SPARK-57268), part of a proposal to make Spark better on a single machine: the first time Arrow shows up as a storage format inside Spark itself rather than at a door. The other goes around the boundary instead of through it: experimental transpilation of Python UDFs into Spark expressions (SPIP SPARK-54783), so a simple enough Python function never has to cross into Python at all. It's off by default, classic-mode only, and requires ANSI mode, but it's the one approach that could get under my benchmark's ceiling, because it removes the per-row Python call rather than making the trip to it cheaper. The pandas side is moving too, more slowly than you'd expect. pandas 3.0, released in January 2026, made a PyArrow-backed string type the default, which should shrink exactly the string-conversion cost that separates pandas UDFs from Arrow UDFs. But PySpark doesn't support pandas 3 yet, not even on the 4.3 branch: it still pins pandas below 3.0, and the umbrella ticket for supporting it (SPARK-55139) is open. When Wes McKinney filed SPARK-13534 in 2016, he was asking for one method to stop pickling. Ten years later, every door he could have named uses Arrow, the defaults finally say so, and the conversion that remains is the one Spark performs on itself, between its own rows and everything outside it. That's the last boundary, and it was never a Python problem.
Most PySpark Tuning Advice Starts in the Wrong Place When I benchmarked PySpark's UDF types earlier this year, I ran the same three workloads (trivial arithmetic, a string transform, a normal CDF) at five scales, from 100 thousand rows to 100 million. At 100 million rows, the default row-at-a-time Python UDF took about 10 seconds for arithmetic that a built-in function finished in under a fifth of a second. At 100 thousand rows, the same UDF was the fastest Python option I tested. Both results are true, and between them they're most of what you need to know about PySpark performance: the cost is almost never in your function. It's in how many times, and in what shape, your data crosses from the JVM into Python, and whether that crossing is large enough to matter at all. My argument is that there is really one PySpark performance technique, applied at different places: cross the JVM/Python boundary less often, move less across it, or stop paying for a JVM where you don't need one. The changes worth making are ordered by how much of that crossing they remove, and the biggest one, not crossing at all, is also the most often skipped. The history of how Arrow came to carry that crossing is in Arrow in PySpark, 2016 to 2026, and the full benchmark is in Stop Blaming Serialization for Slow Python UDFs. This post is the practical version: what to change, in what order, and what each change actually removes. --- Where the time goes PySpark is a Python client for a JVM engine. DataFrame operations build a plan in Python (through Py4J, which only ever carries control messages, never bulk data) and the JVM executes it; df.filter(col("x") > 5).groupBy("y").count() never puts a row into Python, which is why it runs at the same speed as Scala. The data only leaves the JVM when your own code needs it: !diagram 1 Everything outside arrows ① and ② happens at native speed. Every change below makes those two arrows smaller, rarer, or unnecessary, and I've ordered them by how much they took off the arrows in what I measured, or, where I haven't measured, by what they remove. Don't cross at all: built-ins and SQL UDFs The largest win isn't a better UDF; it's no UDF. Built-in functions run inside the JVM as Catalyst expressions, fused by whole-stage code generation into tight loops, and never serialize anything. What surprised me more is that SQL UDFs get the same treatment. Spark 4.0 added SQL user-defined functions (SPARK-46057), and their bodies are inlined into the logical plan, so the optimizer can see straight through them: CREATE FUNCTION classify(score INT) RETURNS STRING RETURN CASE WHEN score > 90 THEN 'excellent' WHEN score > 70 THEN 'good' ELSE 'needs_improvement' END; In my benchmark (median wall-clock time, 50M rows, Spark 4.1.0 through Spark Connect on one machine), the SQL UDF was indistinguishable from the built-ins, and every Python option cost several times more: | | arithmetic | string | normal CDF | |---|---|---|---| | built-in functions | 0.107 s | 0.486 s | 0.176 s | | SQL UDF | 0.095 s | 0.477 s | n/a | | fastest Python UDF (@arrow_udf) | 0.725 s | 1.545 s | 0.762 s | | default Python UDF (@udf, pickle) | 5.199 s | 6.216 s | 6.183 s | So the first question for any UDF is whether it needs to be Python. Plenty of "we need a UDF" logic is a CASE expression, a string function, or a date calculation that already exists in pyspark.sql.functions. Python earns its place when you need a library (scikit-learn, a parser, a C extension) or logic that genuinely can't be expressed in SQL. Everything after this point is about paying less for that choice. Cross once per batch, not once per row If it has to be Python, the next biggest lever is how often Spark calls your function. A regular @udf is called once per row; @pandas_udf and @arrow_udf are called once per batch of rows. That difference showed up as a hard ceiling in my runs: every row-at-a-time UDF, whatever it computed, topped out at roughly 8 to 10 million rows per second, which is roughly the rate at which CPython can be entered and exited. The batched UDFs weren't bound by it. | 50M rows | arithmetic | string | normal CDF | |---|---|---|---| | @udf (pickle, per row) | 5.199 s | 6.216 s | 6.183 s | | @udf(useArrow=True) (Arrow transport, still per row) | 2.228 s | 4.430 s | 2.942 s | | @pandas_udf (per batch, pandas.Series) | 0.789 s | 2.512 s | 0.819 s | | @arrow_udf (per batch, pyarrow.Array) | 0.725 s | 1.545 s | 0.762 s | Two things in that table decide what to change. Moving a per-row UDF onto Arrow transport helps, 1.4x to 2.4x here, and on Spark 4.2 you get it for free: 4.2 finally flipped spark.sql.execution.pythonUDF.arrow.enabled (and its UDTF counterpart) to true by default, after the same change was merged for 4.0 and reverted before release (SPARK-48516). But rewriting to a batched UDF is worth several times more again, because it removes the per-row call rather than making the trip to it cheaper. Between the two batched options, the choice mostly comes down to strings: @arrow_udf was only 5 to 9% faster than @pandas_udf on numeric work, but about 1.6x faster on the string transform, because pandas has traditionally stored strings as individual Python objects while Arrow keeps them in one contiguous buffer. import pyarrow as pa import pyarrow.compute as pc from pyspark.sql.functions import arrow_udf @arrow_udf("string") def add_suffix(x: pa.Array) -> pa.Array: return pc.binary_join_element_wise(pc.utf8_upper(x), pa.scalar("_SUFFIX"), "") One warning, because I learned it the expensive way: spark.sql.execution.arrow.pyspark.enabled is not the UDF setting. It governs toPandas() and createDataFrame(), and it's the one my benchmark originally toggled for its "Arrow" case, which is why I once reported that Arrow transport changes nothing for UDFs. It was the same pickled UDF, measured twice. (On 4.2 that setting is also on by default, so the confusion matters less for new code than for anyone reading old tuning advice.) And the scale caveat from the opening is real. At small sizes, fixed costs dominate, and the ordering can flip: | arithmetic, median seconds | built-in | @pandas_udf | @udf(useArrow=True) | @udf (pickle) | |---|---|---|---|---| | 100K rows | 0.073 | 0.181 | 0.212 | 0.161 | | 10M rows | 0.074 | 0.299 | 0.568 | 1.092 | | 100M rows | 0.147 | 1.400 | 4.188 | 10.139 | If your job touches a few hundred thousand rows, rewriting its UDFs is unlikely to be what makes it faster. Don't produce the data: Python data sources with pushdown After calling Python less, the next thing to remove is data that never needed to exist. Built-in sources like Parquet already push filters down and skip row groups that can't match; for most of Spark's history, custom sources that could do the same had to be written in Java or Scala. Spark 4.0 added the Python Data Source API (SPARK-44076), and Spark 4.1 added filter pushdown for it (SPARK-51271), so a source written in Python can now turn WHERE status = 'active' into a query parameter, a database predicate, or a skipped file block, and never send the rest. When I went back over my own notes on this feature to write this post, the example I'd written was wrong, and checking it against Spark's source turned up something I hadn't known: pushdown for Python sources is off by default. It's governed by spark.sql.python.filterPushdown.enabled, an internal setting added in 4.1 whose description says enabling it costs "running Python worker one additional time during planning"; without it, your pushFilters() is never called. The method also has a less obvious contract than I'd assumed. It returns the filters Spark must still apply after the scan, not the ones you accepted, and each filter's attribute is a column-path tuple: from pyspark.sql.datasource import DataSourceReader, EqualTo spark.conf.set("spark.sql.python.filterPushdown.enabled", "true") # off by default class ApiReader(DataSourceReader): def __init__(self, options): self.url = options.get("url") self.pushed = [] def pushFilters(self, filters): # Filters arrive ANDed together; yield the ones Spark must still apply. for f in filters: if isinstance(f, EqualTo) and f.attribute == ("status",): self.pushed.append(f) # handled at the source else: yield f def read(self, partition): import requests params = {f.attribute[0]: f.value for f in self.pushed} for row in requests.get(self.url, params=params).json(): yield (row["id"], row["name"], row["status"]) Only conjunctions are pushed (a filter like a = 1 or b = 2 arrives as an empty list), and a reader can also yield pyarrow.RecordBatch objects instead of tuples, which the Spark docs say can improve performance by "up to one order of magnitude." I haven't measured either path, so this is the one change on the list I'm ranking by what it removes rather than by what I saw: for a source backed by an API or a database, rows that are never fetched are the cheapest rows there are. Skip pandas for grouped work Per-group logic (normalize within each location, fit a model per customer) has long meant applyInPandas, which converts every group from Arrow into a pandas DataFrame and back. Spark 4.0 added applyInArrow (SPARK-40559), and each group now reaches your function as a pyarrow.Table, in the format it crossed the boundary in: def normalize(table: pa.Table) -> pa.Table: totals = table.column("total") mean, std = pc.mean(totals).as_py(), pc.stddev(totals).as_py() normalized = (pa.array([0.0] * len(totals)) if not std else pc.divide(pc.subtract(totals, mean), std)) return pa.Table.from_arrays([table.column("order_id"), totals, normalized], names=["order_id", "total", "normalized"]) result = orders.groupBy("location_id").applyInArrow( normalize, schema="order_id long, total double, normalized double") This is the same trade as step 2, one level up: it removes a conversion, not a crossing, so the payoff should track what that conversion cost in my UDF numbers (little for numeric groups, more for string-heavy ones, more for larger groups). I haven't benchmarked the grouped APIs, so I won't put a number on it. Take the JVM out of the client Spark Connect is on this list for a different reason than the others. It doesn't make a query faster; it removes the JVM from every client process. Classic PySpark embeds a full Spark driver in each notebook, scheduled job, and CI container. With Connect, the client sends unresolved plans over gRPC and gets Arrow batches back, and the thin pyspark-client package has no JVM and needs no JDK. When I measured it for my own Connect demos on 4.2, the numbers were less dramatic than the pitch: | | pyspark (full) | pyspark-client | |---|---|---| | PyPI artifact | 429.28 MB | 1.60 MB | | installed pyspark/ | 481 MB | 19 MB | | Spark JARs on the client | 456 MB, 276 files | none | | virtual env with Connect dependencies | 820 MB | 359 MB | The last row is the honest one: pyarrow, pandas, and numpy are required dependencies, so the environment shrinks by less than half. What you actually get is a client with no JVM to start, no JDK to install, and no way for a crashed notebook to take the driver down with it. What you give up is anything that needs the JVM on your side: RDDs, SparkContext, custom accumulators. ML on Connect has been supported since 4.0 and generally available since 4.1, so that's no longer on the list of reasons to stay classic. Give Python the memory it actually uses The last change is the one that turns slow jobs into failed ones when it's wrong. Python workers live outside the JVM heap, so the settings most Spark tuning advice is about don't cover them: !diagram 2 The rule is that the container has to fit the heap, the overhead, and the OS, and that Python's share has to come out of the overhead. If you use toPandas() or pandas UDFs, spark.sql.execution.arrow.pyspark.selfDestruct.enabled frees Arrow buffers as they're converted, so the same data doesn't sit in memory twice. Beyond that, the usual advice still holds: let Adaptive Query Execution size shuffle partitions at runtime (it's been on by default since 3.2, and 4.1 extended it to stateless streaming, SPARK-53941, which can change a streaming query's behaviour after an upgrade), and broadcast small dimension tables instead of shuffling them. None of that is specific to Python, which is why it's last on a list about Python. --- What's coming in 4.3 The 4.3 branch hasn't been released (its first release candidate failed its vote on September 6, and there's no new date), but it shows where the boundary is heading, and it confirms the order above. The most interesting change goes around the boundary instead of through it: experimental transpilation of Python UDFs into Spark expressions (SPIP SPARK-54783, spark.sql.experimental.optimizer.transpilePyUDFs). A simple enough Python function would never cross into Python at all, which is change number 1 done for you. It's off by default, requires ANSI mode, and only works in classic mode, not Connect, but it's the only approach on the branch that could get under the per-row ceiling rather than lowering the cost of hitting it. Alongside it are smaller changes to the crossing itself: pipelined, asynchronous transfer of UDF batches (spark.python.udf.pipelined.enabled, SPARK-56642, off by default) and a cap on bytes per UDF batch (SPARK-57593). A proposal to make Spark better on a single machine also adds an Arrow format for Spark's in-memory cache (SPARK-57268). Two things haven't moved. PySpark still doesn't support pandas 3, even on the 4.3 branch; it pins pandas below 3.0, and the ticket to support it (SPARK-55139) is open, so the PyArrow-backed string type pandas 3 made the default in January won't help pandas UDFs yet. And Real-Time Mode in 4.3 gains stateful operators, but still can't run Python UDFs (SPARK-57237 is open). Housekeeping to plan for: 4.2 raised the minimum PyArrow to 18.0.0 and dropped PyPy, and 4.3 drops Python 3.10 (SPARK-56889). --- I opened with a UDF that took 10 seconds at 100 million rows and was the fastest Python option at 100 thousand. The reason is the same one behind every change on this list: your function was never the expensive part. The crossing was, and the best PySpark optimization is still the one that decides the data doesn't need to cross.
Apache Spark won't melt down when your data doesn't fit in RAM anymore Hey, before I start I need to make an honest clarification, because in this series transparency is sacred: the automated analysis I was handed for this post talked about a "memory layer for AI agents," vector DBs, and cross-session context persistence. That has nothing to do with what Apache Spark actually is. That description belongs to a different tool — probably something like Mem0 or similar — and it got crossed wrong in the curation pipeline. It happens, automated systems sometimes mix up metadata. That's why the human verdict still says "pending" — here it is, done by hand, with the real data. What I do have confirmed: Apache Spark shows up in 3 independent awesome lists, and the official repo description describes it as a framework with "micro-batch processing for streams" and "stateful exactly-once semantics" as a backend. In plain terms: it's a distributed engine built for workloads where a single server, no matter how much RAM you throw at it, becomes a bottleneck. Picture this: you have logs from an app generating 500 GB per day, and you need to calculate aggregate metrics — unique users, average latencies, anomaly detection — without waiting 14 hours for a Python script running on a single core to finish. That's where Spark stops being an academic curiosity and becomes the tool that saves your sprint. What it does Apache Spark is a distributed data processing engine, originally written in Scala, that runs on the JVM (yes, the JVM again — there's a reason I keep mentioning it in this series). It was born in Berkeley's AMPLab as a direct response to Hadoop MapReduce's limitations: Spark processes in memory instead of writing to disk at every intermediate step, which makes it orders of magnitude faster for iterative workloads (think ML algorithms, which need to pass over the same data again and again). The central concept is the RDD (Resilient Distributed Dataset): a collection of data partitioned across the cluster's nodes, immutable, and able to rebuild itself if a node goes down (hence "resilient"). On top of that sit friendlier layers like DataFrames and Datasets, which give you a SQL/Pandas-style API, but distributed. // Classic example: counting words in a giant dataset // distributed across all nodes in the cluster import org.apache.spark.sql.SparkSession val spark = SparkSession.builder() .appName("ContadorDePalabras") .getOrCreate() val textFile = spark.read.textFile("hdfs://logs/*.txt") val conteo = textFile .flatMap(linea => linea.split(" ")) // split into words .groupByKey(identity) // group by word .count() // count occurrences conteo.show() spark.stop() The interesting part is that this exact same code runs the same way on your notebook with 4 GB of RAM as it does on a 200-node cluster on AWS EMR. Spark takes care of partitioning the data, distributing the tasks, and handling failures — you just describe what you want to do, not how to distribute it. The streaming part (the one the official repo description mentioned, "micro-batch processing for streams... stateful exactly-once semantics") is Spark Structured Streaming: instead of processing event by event like Kafka Streams or Flink, Spark groups events into micro-batches (every few seconds) and processes them with the same API you use for batch. That gives you "exactly-once" — no event gets processed twice or lost, even if a node explodes at the worst possible moment. # Streaming in PySpark: counting events by time window # reading from a Kafka topic in real time df = spark.readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "localhost:9092") \ .option("subscribe", "eventos") \ .load() conteoPorVentana = df \ .groupBy(window(df.timestamp, "1 minute")) \ .count() query = conteoPorVentana.writeStream \ .outputMode("update") \ .format("console") \ .start() query.awaitTermination() The project is open source under the Apache 2.0 license, maintained by the Apache Software Foundation, with official support for Scala, Java, Python (PySpark), and R. The repo is at apache/spark. Why it's on the list Showing up in 3 independent awesome lists is a decent signal — different curators, different angles, pointing at the same tool. It doesn't prove consensus across entire industries, but it's more than "trending on Twitter": it suggests the tool holds up across more than one context. What sets it apart from alternatives like Hadoop MapReduce (its direct predecessor, much slower because it writes to disk at every step) or Dask (more pythonic but with an ecosystem and maturity well below) is the combination of three things: in-memory speed, a unified API that serves both batch and streaming, and an ecosystem of integrated libraries (MLlib for distributed machine learning, GraphX for graphs, Spark SQL for database-style queries) that saves you from having to glue together five different tools. If you're coming from my history with infrastructure — Linux, networking, servers — you'll quickly understand why Spark earns my respect: it's not magic, it's distributed systems engineering done right, with two decades of academic papers behind it (the original RDD paper by Zaharia et al. is required reading if you like understanding the why, not just the what). When NOT to use it Now for the honest part, because in this series we don't sell smoke: if your dataset fits comfortably in a single machine's RAM (say, under 10-20 GB depending on the hardware), setting up a Spark cluster is like using a crane to lift a soda can. Pandas, Polars, or even DuckDB will give you faster results with a fraction of the operational complexity. It's also not the best option if you need real sub-second latency in streaming — there, Apache Flink beats it hands down because it processes event by event, not in micro-batches. And if your team doesn't have experience operating distributed clusters (YARN, Kubernetes, or Databricks as a managed layer), the maintenance cost can eat up any performance gain. Spark isn't "install and go": it requires understanding partitioning, shuffle, executor memory — things you learn the hard way. Wrap-up This is post #13 of "Awesome Curated: The Tools," the series where we break down tools that passed the filter of our curation system — cross-referencing awesome lists, AI analysis, and a final human verdict. Spark is one of those tools that doesn't have the shine of something new, but that's still there, carrying the real weight of the industry, long after the hype moved on to something else. If you're interested in the infrastructure and systems side of things, check out the post on Node.js and the runtime that changed the backend or the one on Sniffnet for monitoring your network. And if you want to see the full arc of the series, the complete list is at /blog/series/awesome-curated-tools. --- This article was originally published on juanchi.dev