AWS Step Functions Derinlemesine: Dayanıklı Workflow Orchestration Geliştirme

Step Functions ile production serverless workflow kur: Standard ve Express, Distributed Map, error handling ve CDK örnekleriyle maliyet optimizasyonu.

Ayhan Sipahi 4 Aralık 2025

AWS Step Functions orchestration mantığını Lambda kodundan çıkarıp bir state machine’e taşıyor ama sonrasındaki her şeyi iki karar belirliyor: Standard mı Express workflow ve error handling’in ne kadarının state machine’e taşınacağı. Dakikalar içinde biten yüksek hacimli işler için doğru varsayılan Express. Standard’ın daha yüksek transition fiyatı ise workflow günlerce sürüyorsa, exactly-once semantiği gerekiyorsa veya audit trail bırakması gerekiyorsa kendini ödüyor.

Değerin büyük kısmı bu iki kararı doğru vermekte. Geri kalanı production’da ayakta duran küçük bir pattern kümesi: normal Map state’ine sığmayan veri setleri için Distributed Map, human approval’lar için Task Token callback’leri, Lambda wrapper’ını atlayan direct service integration’lar ve deploy etmeden önce hesaplayabileceğiniz bir maliyet modeli.

Orchestration Zorluğu#

Sadece Lambda ile karmaşık serverless workflow’lar geliştirmek bakım sorunları yaratıyor. Lambda fonksiyonlarının içine gömülen orchestration logic hızla şişiyor: yüzlerce satır retry, error handling, state tracking ve conditional branching yönetimi. Production sorunlarını debug etmek, execution flow’ları yeniden oluşturmak için CloudWatch log’larını parse etmeyi gerektiriyor. Yeni adımlar eklemek kod değişikliği ve yeniden deployment gerektiriyor.

Asıl karmaşıklık şu durumlarda ortaya çıkıyor:

Multi-step process’ler: Validation, payment, inventory update ve shipping koordinasyonu içeren sipariş işleme
Error recovery: Uygulama kodunda exponential backoff, circuit breaker ve compensating transaction implementasyonu
State management: Lambda invocation’lar arasında DynamoDB veya Redis kullanarak workflow state’i track etmek latency ve maliyet ekliyor
Parallel processing: Concurrent task’ları koordine ederken failure’ları yönetmek ve sonuçları aggregate etmek
Human approval’lar: Uzun süren approval process’leri için callback mekanizmaları implementasyonu
Scale: Milyonlarca item’ı process etmek doğru orchestration olmadan Lambda timeout limitine takılıyor

Step Functions bu zorlukları visual workflow’lar, built-in error handling ve AWS service integration’larla çözüyor. Ama Standard ve Express workflow arasında seçim yapmak, pricing etkilerini anlamak ve production pattern’lerini uygulamak birden fazla kaynağa yayılmış dokümantasyonda gezinmeyi gerektiriyor.

Workflow Türleri#

Step Functions için temel karar Standard ve Express workflow arasında seçim yapmak. Bu seçim maliyet, execution model ve görünürlüğü etkiliyor.

Standard Workflow’lar#

Standard workflow’lar tam execution history ile exactly-once execution garantisi sağlıyor:

import * as cdk from 'aws-cdk-lib';
import * as sfn from 'aws-cdk-lib/aws-stepfunctions';
import * as tasks from 'aws-cdk-lib/aws-stepfunctions-tasks';

const processOrder = new tasks.LambdaInvoke(this, 'ProcessOrder', {
  lambdaFunction: orderFunction,
  outputPath: '$.Payload'
});

const validateInventory = new tasks.LambdaInvoke(this, 'ValidateInventory', {
  lambdaFunction: inventoryFunction,
  outputPath: '$.Payload'
});

const chargePayment = new tasks.LambdaInvoke(this, 'ChargePayment', {
  lambdaFunction: paymentFunction,
  outputPath: '$.Payload'
});

// Sipariş işleme için Standard workflow
const standardWorkflow = new sfn.StateMachine(this, 'OrderProcessing', {
  stateMachineType: sfn.StateMachineType.STANDARD,
  definition: processOrder
    .next(validateInventory)
    .next(chargePayment),
  timeout: cdk.Duration.days(7)
});

Standard workflow özellikleri:

Maksimum süre: 1 yıl (multi-day approval process’leri için kullanışlı)
Exactly-once workflow execution, Retry policy verdiğiniz task’lar hariç
90 gün boyunca tam execution history saklanıyor
Fiyatlandırma: 1.000 state transition başına $0.025
Step Functions console’da tam görünürlük
.sync ve .waitForTaskToken integration pattern’lerini destekliyor

Express Workflow’lar#

Express workflow’lar high-throughput, kısa süreli processing için optimize edilmiş:

const receiveIoTEvent = new tasks.LambdaInvoke(this, 'ReceiveEvent', {
  lambdaFunction: receiveFunction,
  outputPath: '$.Payload'
});

const validateData = new tasks.LambdaInvoke(this, 'ValidateData', {
  lambdaFunction: validateFunction,
  outputPath: '$.Payload'
});

const storeData = new tasks.DynamoPutItem(this, 'StoreData', {
  table: dataTable,
  item: {
    id: tasks.DynamoAttributeValue.fromString(
      sfn.JsonPath.stringAt('$.eventId')
    ),
    timestamp: tasks.DynamoAttributeValue.fromNumber(
      sfn.JsonPath.numberAt('$.timestamp')
    )
  }
});

// IoT processing için Express workflow
const expressWorkflow = new sfn.StateMachine(this, 'IoTProcessing', {
  stateMachineType: sfn.StateMachineType.EXPRESS,
  definition: receiveIoTEvent
    .next(validateData)
    .next(storeData),
  timeout: cdk.Duration.minutes(5)
});

Express workflow özellikleri:

Maksimum süre: 5 dakika
At-least-once execution (birden fazla kez çalışabilir)
Sınırlı execution history (sadece CloudWatch Logs)
Fiyatlandırma: 1M request başına $1.00 + GB-second başına $0.00001667
Throughput: Saniyede 100.000+ execution
İki mod: Synchronous (sonucu bekle) ve Asynchronous (fire-and-forget)

Maliyet Karşılaştırması#

Fiyatlandırma farkı scale’de belirgin hale geliyor:

// Senaryo: Ayda 10 milyon execution
// Her workflow: 10 state transition
// Ortalama süre: 2 saniye
// Memory: 512MB

// Standard Workflow:
// Toplam transition'lar: 10M * 10 = 100M
// Maliyet = (100.000.000 / 1.000) * $0.025 = $2.500/ay

// Express Workflow:
// Request maliyeti: (10.000.000 / 1.000.000) * $1.00 = $10
// GB-saniye: (512/1024) * 2 * 10.000.000 = 10.000.000
// Süre maliyeti: 10.000.000 * $0.00001667 = $167
// Toplam: $177/ay

// Tasarruf: $2.323/ay (%93 azalma)

Yüksek hacimli, kısa süreli workflow’lar için Express workflow’lar önemli maliyet tasarrufu sağlıyor. Standard workflow’lar uzun süren process’ler, exactly-once gereksinimi veya audit trail ihtiyacı için mantıklı.

Amazon States Language Pattern’leri#

Step Functions workflow’ları JSON tabanlı bir spesifikasyon olan Amazon States Language (ASL) kullanılarak tanımlanıyor. Data flow kontrolünü anlamak, maintainable workflow’lar kurmanın ön koşulu.

Data Flow Kontrolü#

ASL, workflow state’leri arasında data akışını kontrol etmek için birkaç mekanizma sağlıyor:

// Conditional routing ile Choice state
const decisionTree = new sfn.Choice(this, 'RouteByWorkload', {
  stateName: 'Workload Router'
})
  .when(
    sfn.Condition.numberGreaterThan('$.itemCount', 1000),
    largeBatchProcessing
  )
  .when(
    sfn.Condition.numberGreaterThan('$.itemCount', 100),
    mediumBatchProcessing
  )
  .otherwise(smallBatchProcessing);

// Parallel processing için Map state
const processItems = new sfn.Map(this, 'ProcessEachItem', {
  maxConcurrency: 10,
  itemsPath: '$.items',
  parameters: {
    'item.$': '$$.Map.Item.Value',
    'index.$': '$$.Map.Item.Index',
    'executionId.$': '$$.Execution.Id'
  }
}).itemProcessor(
  new sfn.Pass(this, 'ProcessItem')
);

// Simultaneous execution için Parallel state
const parallelProcessing = new sfn.Parallel(this, 'FanOut', {
  resultPath: '$.parallelResults'
})
  .branch(processPayment)
  .branch(updateInventory)
  .branch(sendNotification);

// Dynamic timestamp ile Wait state
const waitForSchedule = new sfn.Wait(this, 'WaitUntilScheduled', {
  time: sfn.WaitTime.timestampPath('$.scheduledTime')
});

ResultPath ve Data Transformation#

ResultPath parametresi task output’unun state output’unda nereye yerleştirileceğini kontrol ediyor:

// Output'u filter etmek için ResultSelector ile task
const transformedTask = new tasks.LambdaInvoke(this, 'GetUserData', {
  lambdaFunction: getUserFunction,
  resultSelector: {
    'userId.$': '$.Payload.id',
    'userName.$': '$.Payload.name',
    'email.$': '$.Payload.email'
  },
  resultPath: '$.user'
});

// Input: { orderId: '123', customerId: '456' }
// Lambda dönüyor: { id: 'u789', name: 'John', email: 'john@example.com', internalData: {...} }
// ResultSelector filter ediyor: { userId: 'u789', userName: 'John', email: 'john@example.com' }
// ResultPath yerleştiriyor: { orderId: '123', customerId: '456', user: { userId: 'u789', ... } }

Context Object Variable’ları#

ASL execution metadata’ya erişmek için context variable’ları sağlıyor:

const taskWithContext = new tasks.LambdaInvoke(this, 'ProcessWithContext', {
  lambdaFunction: processFunction,
  payload: sfn.TaskInput.fromObject({
    'data.$': '$.inputData',
    'executionId.$': '$$.Execution.Id',
    'executionName.$': '$$.Execution.Name',
    'stateMachineId.$': '$$.StateMachine.Id',
    'timestamp.$': '$$.State.EnteredTime'
  })
});

Error Handling ve Retry Stratejileri#

Production workflow’lar kapsamlı error handling gerektiriyor. Step Functions built-in retry ve catch mekanizmaları sağlıyor.

Retry ile Exponential Backoff#

// Retry'lar task'ı yeniden invoke ediyor: paymentFunction order ID üzerinden idempotent olmalı
const resilientTask = new tasks.LambdaInvoke(this, 'ProcessPayment', {
  lambdaFunction: paymentFunction,
  payloadResponseOnly: true,
  retryOnServiceExceptions: true
})
  .addRetry({
    // Transient error'lar için hızlı retry'lar
    errors: ['States.TaskFailed', 'ThrottlingException', 'ServiceUnavailable'],
    interval: cdk.Duration.seconds(2),
    maxAttempts: 3,
    backoffRate: 2.0 // 2s, 4s, 8s
  })
  .addRetry({
    // Timeout error'ları için farklı strateji
    errors: ['States.Timeout'],
    interval: cdk.Duration.seconds(5),
    maxAttempts: 2,
    backoffRate: 1.5 // 5s, 7.5s
  });

Retry mekanizması transient failure’ları ek kod olmadan handle ediyor. backoffRate parametresi exponential backoff’u kontrol ediyor: 2.0 oranı her retry’dan sonra bekleme süresini ikiye katlıyor.

Retry ikinci bir invocation, kaldığı yerden devam etme değil; yukarıdaki States.Timeout retry’ı da tam burada canınızı yakıyor. Handler kartı çekmiş ama success’i bildirmeden timeout’a düşmüş olabilir. Sonraki denemenin bu çekimi tekrar etmesi değil, idempotency key’inden tanıması gerekiyor. Standard workflow’lar bunu değiştirmiyor: exactly-once modeli, Retry policy verdiğiniz task’ları kapsam dışında bırakıyor.

Not: payloadResponseOnly: true response yapısını tam Step Functions wrapper yerine sadece Lambda payload’ını döndürerek basitleştiriyor. Ancak bu workflow state’inde StatusCode ve ExecutedVersion gibi metadata’ya erişimi kaybettiğiniz anlamına geliyor. Bu metadata’ya debugging veya auditing için ihtiyacınız varsa bunun yerine outputPath: '$.Payload' kullanın.

Error Catching ve Compensation#

const handlePaymentFailure = new tasks.SqsSendMessage(this, 'NotifyFailure', {
  queue: failureQueue,
  messageBody: sfn.TaskInput.fromObject({
    'orderId.$': '$.orderId',
    'error.$': '$.errorInfo.Error',
    'cause.$': '$.errorInfo.Cause'
  })
});

const notifyAdmin = new tasks.SnsPublish(this, 'AlertAdmin', {
  topic: adminTopic,
  message: sfn.TaskInput.fromText('Critical payment processing failure')
});

const paymentTask = new tasks.LambdaInvoke(this, 'ChargeCustomer', {
  lambdaFunction: paymentFunction,
  payloadResponseOnly: true
})
  .addCatch(handlePaymentFailure, {
    // Specific business error'ları handle et
    errors: ['PaymentDeclined', 'InsufficientFunds'],
    resultPath: '$.errorInfo'
  })
  .addCatch(notifyAdmin, {
    // Diğer tüm error'ları yakala
    errors: ['States.ALL'],
    resultPath: '$.error'
  });

Catch block’larındaki resultPath original input’u korurken error bilgisini ekliyor. Bu downstream state’lerin hem input data’ya hem error detaylarına erişmesini sağlıyor.

Lambda Error Handling#

Lambda fonksiyonları Step Functions’ın yakalayabileceği specific error type’ları throw etmeli:

export const handler = async (event: { orderId: string, amount: number }) => {
  try {
    const result = await processPayment(event.amount, event.orderId);
    return result;
  } catch (error: any) {
    // Step Functions'ın yakalayabileceği specific error type'ları throw et
    if (error.code === 'INSUFFICIENT_FUNDS') {
      throw new Error('InsufficientFunds');
    }
    if (error.code === 'CARD_DECLINED') {
      throw new Error('PaymentDeclined');
    }
    // Generic retry logic için yeniden throw et
    throw error;
  }
};

Circuit Breaker Pattern#

Birden fazla fallback seçeneği olan sistemler için:

// Önce fallback'leri tanımla: her addCatch hedefinin önceden var olması gerekiyor
const logFailure = new tasks.DynamoPutItem(this, 'LogFailure', {
  table: failureTable,
  item: {
    id: tasks.DynamoAttributeValue.fromString(
      sfn.JsonPath.stringAt('$.requestId')
    ),
    // $$.State.EnteredTime bir ISO-8601 metni, epoch sayısı değil
    timestamp: tasks.DynamoAttributeValue.fromString(
      sfn.JsonPath.stringAt('$$.State.EnteredTime')
    )
  }
});

const fallbackApi = new tasks.LambdaInvoke(this, 'FallbackAPI', {
  lambdaFunction: fallbackApiFunction,
  payloadResponseOnly: true
})
  .addCatch(logFailure, {
    errors: ['States.ALL'],
    resultPath: '$.fallbackError'
  });

const primaryApi = new tasks.LambdaInvoke(this, 'PrimaryAPI', {
  lambdaFunction: primaryApiFunction,
  payloadResponseOnly: true
})
  .addCatch(fallbackApi, {
    errors: ['States.ALL'],
    resultPath: '$.primaryError'
  });

Bu pattern her seviyede error context korunarak otomatik fallback sağlıyor.

Büyük Ölçekli Processing için Distributed Map#

Normal Map state’leri 40’a kadar concurrent iteration’ı destekliyor. Distributed Map milyonlarca item’ı process etmek için bu sınırı kaldırıyor.

Temel Distributed Map#

import * as s3 from 'aws-cdk-lib/aws-s3';

const inputBucket = s3.Bucket.fromBucketName(this, 'InputBucket', 'data-input');
const resultsBucket = s3.Bucket.fromBucketName(this, 'ResultsBucket', 'data-results');

const processRecord = new tasks.LambdaInvoke(this, 'ProcessRecord', {
  lambdaFunction: processFunction,
  payloadResponseOnly: true
});

const distributedMap = new sfn.DistributedMap(this, 'ProcessLargeDataset', {
  maxConcurrency: 10000,
  itemReader: new sfn.S3JsonItemReader({
    bucket: inputBucket,
    key: 'input-data/*.json'
  }),
  resultWriter: new sfn.ResultWriter({
    bucket: resultsBucket,
    prefix: 'processing-results/'
  }),
  toleratedFailurePercentage: 5
});

distributedMap.itemProcessor(processRecord);

toleratedFailurePercentage parametresi bazı item’lar fail olsa bile processing’e devam etmeye izin veriyor. Bu partial failure’ların kabul edilebilir olduğu batch processing için kullanışlı.

CSV ve JSONL Processing#

Distributed Map birden fazla input formatını destekliyor:

// CSV dosyalarını process et
const csvProcessing = new sfn.DistributedMap(this, 'ProcessCSV', {
  itemReader: new sfn.S3CsvItemReader({
    bucket: csvBucket,
    key: 'data/*.csv',
    csvHeaders: sfn.CsvHeaders.use(['id', 'name', 'value', 'timestamp'])
  })
});

// JSON Lines dosyalarını process et
const jsonlProcessing = new sfn.DistributedMap(this, 'ProcessJSONL', {
  itemReader: new sfn.S3JsonItemReader({
    bucket: logsBucket,
    key: 'application-logs/*.jsonl'
  })
});

Ölçekte Concurrency ve Maliyet#

Distributed Map parayı wall-clock süreyle takas ediyor ve bu takasın iki tarafı da çalıştırmadan önce hesaplanabiliyor:

// Senaryo: S3'te 12.000 JSON dosyasına dağılmış 10 milyon sensor reading

const processReadings = new sfn.DistributedMap(this, 'ProcessSensorData', {
  maxConcurrency: 8000,
  itemReader: new sfn.S3JsonItemReader({
    bucket: sensorBucket,
    key: 'readings/*.json'
  }),
  resultWriter: new sfn.ResultWriter({
    bucket: resultsBucket,
    prefix: 'processed/'
  })
});

// Wall-clock süre item sayısını değil maxConcurrency'yi takip ediyor:
// tek bir sequential döngü yerine 8.000 child execution aynı anda çalışıyor.
// Fatura yine item sayısını takip ediyor: 1.000 state transition başına
// $0.025'ten 10M transition, Lambda ve S3 hariç yaklaşık $250.

maxConcurrency aynı anda kaç child execution’ın çalışacağını belirliyor, tavanı 10.000. Dolayısıyla süre batch sayısıyla ölçekleniyor, fatura ise her hâlükârda item sayısıyla. Maliyet latency’den önemliyse item processor içinde batch’leyin ve concurrency’yi düşürün.

Task Token’larla Callback Pattern#

Task Token’lar workflow’un bir external event’i beklerken duraklamasını sağlıyor; human approval’ları ve uzun süren diğer adımları mümkün kılan da bu.

Human Approval Workflow#

import * as sqs from 'aws-cdk-lib/aws-sqs';

const approvalQueue = new sqs.Queue(this, 'ApprovalQueue');

const requestApproval = new tasks.SqsSendMessage(this, 'SendApprovalRequest', {
  queue: approvalQueue,
  messageBody: sfn.TaskInput.fromObject({
    'orderId.$': '$.orderId',
    'amount.$': '$.amount',
    'customerId.$': '$.customerId',
    'taskToken': sfn.JsonPath.taskToken // Kritik: Callback için task token
  }),
  integrationPattern: sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
  timeout: cdk.Duration.hours(24)
});

const handleTimeout = new tasks.SnsPublish(this, 'NotifyTimeout', {
  topic: timeoutTopic,
  message: sfn.TaskInput.fromText('Approval request timed out')
});

const approvalTask = requestApproval
  .addCatch(handleTimeout, {
    errors: ['States.Timeout'],
    resultPath: '$.timeoutError'
  });

Workflow bu state’te task token ile SendTaskSuccess veya SendTaskFailure çağrılana kadar duraklatılıyor.

Approval Request’leri İşleme#

import { SQSEvent } from 'aws-lambda';
import { DynamoDBClient, PutItemCommand } from '@aws-sdk/client-dynamodb';

const dynamodb = new DynamoDBClient({});

export const approvalHandler = async (event: SQSEvent) => {
  for (const record of event.Records) {
    const { orderId, amount, customerId, taskToken } = JSON.parse(record.body);

    // Admin UI için approval request'i sakla
    await dynamodb.send(new PutItemCommand({
      TableName: process.env.APPROVALS_TABLE!,
      Item: {
        orderId: { S: orderId },
        amount: { N: amount.toString() },
        customerId: { S: customerId },
        taskToken: { S: taskToken },
        status: { S: 'PENDING' },
        timestamp: { N: Date.now().toString() }
      }
    }));
  }
};

Approval Decision Callback#

import { APIGatewayProxyEvent, APIGatewayProxyResult } from 'aws-lambda';
import { SFNClient, SendTaskSuccessCommand, SendTaskFailureCommand } from '@aws-sdk/client-sfn';
import { DynamoDBClient, GetItemCommand } from '@aws-sdk/client-dynamodb';

const sfn = new SFNClient({});
const dynamodb = new DynamoDBClient({});

export const approveHandler = async (event: APIGatewayProxyEvent): Promise<APIGatewayProxyResult> => {
  const { orderId, decision } = JSON.parse(event.body || '{}');

  // Approval request'i al
  const result = await dynamodb.send(new GetItemCommand({
    TableName: process.env.APPROVALS_TABLE!,
    Key: { orderId: { S: orderId } }
  }));

  if (!result.Item) {
    return { statusCode: 404, body: 'Approval request not found' };
  }

  const taskToken = result.Item.taskToken.S!;

  if (decision === 'approved') {
    await sfn.send(new SendTaskSuccessCommand({
      taskToken,
      output: JSON.stringify({ approved: true, orderId })
    }));
  } else {
    await sfn.send(new SendTaskFailureCommand({
      taskToken,
      error: 'ApprovalRejected',
      cause: 'Order rejected by admin'
    }));
  }

  return { statusCode: 200, body: 'Decision recorded' };
};

Callback invoke edildiğinde workflow execution’a devam ediyor. Bu pattern task token’ları saklayıp retrieve edebilen herhangi bir external sistemle çalışıyor.

Direct Service Integration’lar#

Step Functions 220’nin üzerinde AWS servisiyle doğrudan entegre oluyor ve Lambda wrapper’larına duyulan ihtiyacı ortadan kaldırıyor.

DynamoDB Integration#

const saveOrder = new tasks.DynamoPutItem(this, 'SaveOrder', {
  table: ordersTable,
  item: {
    orderId: tasks.DynamoAttributeValue.fromString(
      sfn.JsonPath.stringAt('$.orderId')
    ),
    customerId: tasks.DynamoAttributeValue.fromString(
      sfn.JsonPath.stringAt('$.customerId')
    ),
    amount: tasks.DynamoAttributeValue.fromNumber(
      sfn.JsonPath.numberAt('$.amount')
    ),
    timestamp: tasks.DynamoAttributeValue.fromString(
      sfn.JsonPath.stringAt('$$.State.EnteredTime')
    ),
    status: tasks.DynamoAttributeValue.fromString('PROCESSING')
  }
});

Bu yaklaşım Lambda invocation maliyetlerini azaltıyor ve latency’yi düşürüyor.

const notifyUser = new tasks.SnsPublish(this, 'NotifyUser', {
  topic: notificationTopic,
  message: sfn.TaskInput.fromObject({
    'orderId.$': '$.orderId',
    'status': 'processing',
    'timestamp.$': '$$.State.EnteredTime'
  })
});

const queueWork = new tasks.SqsSendMessage(this, 'QueueWork', {
  queue: workQueue,
  messageBody: sfn.TaskInput.fromJsonPathAt('$.workload'),
  messageDeduplicationId: sfn.JsonPath.stringAt('$.orderId')
});

ECS Task’ı .sync ile Çalıştırma#

Uzun süren batch job’lar için:

import * as ecs from 'aws-cdk-lib/aws-ecs';

const runBatchJob = new tasks.EcsRunTask(this, 'RunDataProcessing', {
  cluster: ecsCluster,
  taskDefinition: batchJobTask,
  launchTarget: new tasks.EcsFargateLaunchTarget(),
  integrationPattern: sfn.IntegrationPattern.RUN_JOB, // .sync pattern
  containerOverrides: [{
    containerDefinition: batchContainer,
    environment: [
      { name: 'INPUT_BUCKET', value: 'data-input' },
      { name: 'OUTPUT_BUCKET', value: 'data-output' }
    ]
  }]
});

Workflow ECS task tamamlanana kadar bekliyor, bu saatler sürebilir. Data processing, ML training veya diğer batch operation’lar için kullanışlı.

SDK Service Integration#

Dedicated CDK construct’ları olmayan servisler için:

const updateSecret = new tasks.CallAwsService(this, 'UpdateSecret', {
  service: 'secretsmanager',
  action: 'updateSecret',
  parameters: {
    SecretId: 'MyApplicationSecret',
    SecretString: sfn.JsonPath.stringAt('$.newSecretValue')
  },
  iamResources: ['arn:aws:secretsmanager:*:*:secret:MyApplicationSecret-*']
});

CallAwsService construct’ı Step Functions’tan doğrudan herhangi bir AWS SDK API’sini çağırmayı sağlıyor.

Production Implementation Pattern#

Logging, monitoring ve error handling içeren tam workflow implementasyonu:

import { Construct } from 'constructs';
import * as cdk from 'aws-cdk-lib';
import * as sfn from 'aws-cdk-lib/aws-stepfunctions';
import * as tasks from 'aws-cdk-lib/aws-stepfunctions-tasks';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as logs from 'aws-cdk-lib/aws-logs';
import * as cloudwatch from 'aws-cdk-lib/aws-cloudwatch';
import * as sns from 'aws-cdk-lib/aws-sns';
import * as cloudwatch_actions from 'aws-cdk-lib/aws-cloudwatch-actions';

export class OrderProcessingWorkflow extends Construct {
  public readonly stateMachine: sfn.StateMachine;

  constructor(scope: Construct, id: string) {
    super(scope, id);

    // Lambda fonksiyonları
    const validateOrder = new lambda.Function(this, 'ValidateOrder', {
      runtime: lambda.Runtime.NODEJS_24_X,
      handler: 'validate.handler',
      code: lambda.Code.fromAsset('lambda'),
      timeout: cdk.Duration.seconds(30)
    });

    const processPayment = new lambda.Function(this, 'ProcessPayment', {
      runtime: lambda.Runtime.NODEJS_24_X,
      handler: 'payment.handler',
      code: lambda.Code.fromAsset('lambda'),
      timeout: cdk.Duration.seconds(60)
    });

    const shipOrder = new lambda.Function(this, 'ShipOrder', {
      runtime: lambda.Runtime.NODEJS_24_X,
      handler: 'shipping.handler',
      code: lambda.Code.fromAsset('lambda'),
      timeout: cdk.Duration.seconds(30)
    });

    // Task tanımları
    const validateTask = new tasks.LambdaInvoke(this, 'ValidateOrderTask', {
      lambdaFunction: validateOrder,
      outputPath: '$.Payload'
    });

    // paymentTask'tan önce tanımlanıyor: addCatch hedefi zaten var olmalı
    const handlePaymentFailure = new sfn.Fail(this, 'PaymentFailed', {
      error: 'PaymentProcessingFailed',
      cause: 'Unable to process payment after retries'
    });

    const paymentTask = new tasks.LambdaInvoke(this, 'ProcessPaymentTask', {
      lambdaFunction: processPayment,
      outputPath: '$.Payload',
      retryOnServiceExceptions: true
    })
      .addRetry({
        errors: ['ThrottlingException'],
        interval: cdk.Duration.seconds(1),
        maxAttempts: 3,
        backoffRate: 2
      })
      .addCatch(handlePaymentFailure, {
        errors: ['PaymentFailed'],
        resultPath: '$.error'
      });

    // Shipping'e yalnızca başarılı ödeme ulaşıyor: yakalanan hata Fail state'inde bitiyor
    const shippingTask = new tasks.LambdaInvoke(this, 'ShipOrderTask', {
      lambdaFunction: shipOrder,
      outputPath: '$.Payload'
    });

    const notifyInvalidOrder = new sfn.Fail(this, 'InvalidOrder', {
      error: 'OrderValidationFailed',
      cause: 'Order validation did not pass'
    });

    // Workflow tanımı
    const definition = validateTask
      .next(new sfn.Choice(this, 'OrderValid?')
        .when(
          sfn.Condition.booleanEquals('$.valid', true),
          paymentTask.next(shippingTask)
        )
        .otherwise(notifyInvalidOrder)
      );

    // CloudWatch log group
    const logGroup = new logs.LogGroup(this, 'Logs', {
      retention: logs.RetentionDays.ONE_WEEK,
      removalPolicy: cdk.RemovalPolicy.DESTROY
    });

    // Logging ve tracing ile state machine
    this.stateMachine = new sfn.StateMachine(this, 'OrderProcessing', {
      definition,
      stateMachineType: sfn.StateMachineType.EXPRESS,
      logs: {
        destination: logGroup,
        level: sfn.LogLevel.ERROR,
        includeExecutionData: true
      },
      tracingEnabled: true // X-Ray'i etkinleştir
    });

    // CloudWatch alarm'ları
    const alertTopic = new sns.Topic(this, 'AlertTopic');

    const failureAlarm = new cloudwatch.Alarm(this, 'HighFailureRate', {
      metric: this.stateMachine.metricFailed({
        period: cdk.Duration.minutes(5)
      }),
      threshold: 10,
      evaluationPeriods: 2,
      treatMissingData: cloudwatch.TreatMissingData.NOT_BREACHING
    });

    failureAlarm.addAlarmAction(new cloudwatch_actions.SnsAction(alertTopic));
  }
}

Bu implementasyon production monitoring için error handling, logging, X-Ray tracing ve CloudWatch alarm’ları içeriyor.

Maliyet Optimizasyon Stratejileri#

Step Functions pricing’i anlamak önemli maliyet azaltmaları sağlıyor.

State Transition Sayısını Azaltma#

// Anti-pattern: Gereksiz Pass state'leri
const inefficient = sfn.Chain.start(task1)
  .next(new sfn.Pass(this, 'PassData1', {})) // Gereksiz
  .next(task2)
  .next(new sfn.Pass(this, 'PassData2', {})) // Gereksiz
  .next(task3);
// Maliyet: 5 state transition

// Optimize edilmiş: Gereksiz state'leri kaldır
const efficient = sfn.Chain.start(task1)
  .next(task2)
  .next(task3);
// Maliyet: 3 state transition (%40 azalma)

Batch Processing#

Item’ları tek tek process etmek çok sayıda state transition yaratıyor. Batching maliyetleri azaltıyor:

// 10.000 item'ı 100'lük 100 batch'te process et
const batchProcessor = new sfn.Map(this, 'ProcessBatches', {
  itemsPath: '$.batches',
  maxConcurrency: 10
}).itemProcessor(
  new tasks.LambdaInvoke(this, 'ProcessBatch', {
    lambdaFunction: batchFunction,
    payload: sfn.TaskInput.fromObject({
      'items.$': '$$.Map.Item.Value',
      'batchId.$': '$$.Map.Item.Index'
    })
  })
);

// Tekli processing: 10.000 state transition
// Batch processing: 100 state transition
// Tasarruf: %99 azalma

Direct Service Integration’lar#

Direct integration’ları kullanmak Lambda maliyetlerini ortadan kaldırıyor:

// Orijinal: DynamoDB için Lambda wrapper
const withLambda = new tasks.LambdaInvoke(this, 'SaveOrder', {
  lambdaFunction: dynamoWrapperFunction
});
// Maliyet: Lambda invocation + state transition

// Optimize edilmiş: Direct DynamoDB integration
const directIntegration = new tasks.DynamoPutItem(this, 'SaveOrder', {
  table: ordersTable,
  item: { /* ... */ }
});
// Maliyet: Sadece state transition (Lambda maliyeti yok)

Monitoring ve Observability#

Production workflow’lar kapsamlı monitoring gerektiriyor.

CloudWatch Metric’leri#

const workflow = new sfn.StateMachine(this, 'ProductionWorkflow', {
  definition,
  tracingEnabled: true
});

// Failure rate alarm'ı
const failureAlarm = new cloudwatch.Alarm(this, 'FailureRate', {
  metric: workflow.metricFailed({
    period: cdk.Duration.minutes(5),
    statistic: cloudwatch.Stats.SUM
  }),
  threshold: 10,
  evaluationPeriods: 2
});

// Duration alarm'ı (p99)
const durationAlarm = new cloudwatch.Alarm(this, 'LongExecution', {
  metric: workflow.metricDuration({
    statistic: cloudwatch.Stats.PERCENTILE_99
  }),
  threshold: cdk.Duration.minutes(10).toMilliseconds(),
  evaluationPeriods: 1
});

// Throttling alarm'ı
const throttleAlarm = new cloudwatch.Alarm(this, 'Throttling', {
  metric: workflow.metricThrottled({
    period: cdk.Duration.minutes(1)
  }),
  threshold: 1,
  evaluationPeriods: 1
});

CloudWatch Dashboard#

const dashboard = new cloudwatch.Dashboard(this, 'WorkflowDashboard', {
  dashboardName: 'step-functions-monitoring'
});

dashboard.addWidgets(
  new cloudwatch.GraphWidget({
    title: 'Execution Rate',
    left: [
      workflow.metricStarted({ label: 'Started' }),
      workflow.metricSucceeded({ label: 'Succeeded' }),
      workflow.metricFailed({ label: 'Failed' })
    ],
    width: 12
  }),
  new cloudwatch.GraphWidget({
    title: 'Execution Duration (ms)',
    left: [
      workflow.metricDuration({
        statistic: cloudwatch.Stats.AVERAGE,
        label: 'Average'
      }),
      workflow.metricDuration({
        statistic: cloudwatch.Stats.PERCENTILE_99,
        label: 'p99'
      })
    ],
    width: 12
  })
);

X-Ray Tracing#

Servis dependency’lerini görselleştirmek ve bottleneck’leri belirlemek için X-Ray’i etkinleştirin:

X-Ray her servis çağrısı için execution time gösteriyor, yavaş component’leri belirlemek kolaylaşıyor.

Alternatif Araçlar#

Step Functions her zaman doğru seçim değil. Alternatifleri ne zaman değerlendirmeli:

Temporal#

Temporal’ı şu durumlarda kullanın:

JSON/CDK tanımları yerine code-as-workflow tercih edildiğinde
Multi-cloud deployment gerektiğinde
Karmaşık business logic’in kendisi workflow’un içinde durduğunda
Saniye altı latency kritik olduğunda
Mock olmadan local development önemli olduğunda

// Temporal workflow (TypeScript SDK)
export async function orderProcessingWorkflow(order: Order): Promise<void> {
  await activities.validateOrder(order);

  try {
    await activities.processPayment(order);
  } catch (err) {
    await activities.refundPayment(order);
    throw err;
  }

  await activities.shipOrder(order);
}

// Workflow logic TypeScript ile yazılır, uygulama kodu ile birlikte versiyonlanır
// Daha iyi IDE desteği, debugging ve test imkanı

AWS MWAA (Managed Airflow)#

Airflow’u şu durumlarda kullanın:

Data pipeline orchestration (ETL, batch processing) gerektiğinde
Scheduled job’lar arasında karmaşık dependency’ler olduğunda
Data engineer’lar için zengin bir arayüz şart olduğunda
Data tool’larıyla (Spark, Hive, Presto) entegrasyon gerektiğinde

MWAA sürekli çalışan bir environment üzerinden faturalandırıyor, Step Functions ise execution başına. Ani yükselen, event-driven workload’larda karşılaştırmayı saatlik ücretten bağımsız olarak bu fark belirliyor.

EventBridge Pipes#

Karmaşık logic olmadan basit event transformation ve routing için kullanın:

const pipe = new pipes.Pipe(this, 'SimpleProcessing', {
  source: new pipes.SqsSource(queue),
  target: new pipes.LambdaTarget(processFunction),
  enrichment: new pipes.LambdaEnrichment(transformFunction),
  filter: pipes.Filter.fromObject({
    body: {
      amount: [{ numeric: ['>', 100] }]
    }
  })
});

Branching veya karmaşık error handling olmadan linear pipeline’lar için daha basit.

Sık Yapılan Hatalar#

Express workflow’lar at-least-once execution garantisi veriyor, dolayısıyla bir execution bütünüyle iki kez çalışabiliyor. Standard workflow’lar bu kapıyı yalnızca execution seviyesinde kapatıyor. Retry policy’si olan bir task yine çalışıyor; yan etkisini tamamlayıp success’i bildirmeden timeout’a düşen bir worker da öyle. İki tür de sizin yerinize deduplication yapmıyor. Hangi workflow türünü seçerseniz seçin, giriş state’inde payload’a bir idempotency key yazın ve kart çekmeden ya da sipariş göndermeden önce handler içinde bunu kontrol edin.

Task token’lar kendiliğinden expire olmuyor. timeout verilmemiş bir WAIT_FOR_TASK_TOKEN task’ı execution’ı state machine’in kendi limitine kadar tutuyor; Standard workflow’larda bu limit bir yıl. Task seviyesinde timeout verin ve approval örneğindeki gibi States.Timeout’u yakalayın.

resultPath belirtilmediğinde task sonucu state input’unun yerine geçiyor. Sonraki state’ler o noktada order ID’yi, correlation ID’yi ve workflow’un taşıdığı diğer her şeyi kaybediyor. Input’una sonraki state’lerin ihtiyaç duyduğu her task’ta resultPath belirtin.

Varsayılanı Seçmek#

Dakikalar içinde biten event-driven işler için doğru varsayılan Express. Express request ve GB-saniye üzerinden fiyatlanıyor, Standard ise her state transition üzerinden; yukarıdaki hesap iki modelin hacim arttıkça ne kadar açıldığını gösteriyor. Bu varsayılanı üç durumda bozun. İnsan ya da external sistem bekleyen uzun süreli process’ler Standard’ın bir yıllık tavanına ve callback pattern’ine ihtiyaç duyuyor. Bütünüyle yeniden çalışmaması gereken işler Standard’ın exactly-once execution’ını istiyor, çünkü Express aynı execution’ı bir kez daha başlatabiliyor. Bu garanti execution sınırında bitiyor; Retry policy verdiğiniz her task’ın yine bir idempotency key’e ihtiyacı var. Audit gereksinimleri Standard’ın 90 günlük execution history’sine bağlı, çünkü Express yalnızca CloudWatch Logs’a gönderdiğinizi saklıyor.

Aynı mantık iki küçük varsayılan daha üretiyor. Bir adım CallAwsService çağrısının ifade edemeyeceği bir logic gerektirmiyorsa, Lambda wrapper yerine direct service integration tercih edin. Veri seti normal Map state’inin 40 concurrent iteration’ını aştığında Distributed Map’e geçin ve faturayı state transition’lar belirliyorsa item’ları iterate etmeden önce batch’leyin.

Kaynaklar#

AWS Step Functions Nedir? (yeni sekmede açılır) - State machine’leri, workflow türlerini ve temel orkestrasyon kavramlarını tanıtan resmi geliştirici kılavuzu
Workflow Türü Seçimi (Standard ve Express) (yeni sekmede açılır) - Tam-bir-kez Standard Workflow’ları ile yüksek hacimli kullanım durumları için en-az-bir-kez Express Workflow’ları karşılaştıran karar kılavuzu
Amazon States Language (ASL) Yapısı (yeni sekmede açılır) - Step Functions state machine’lerini tanımlamak için kullanılan JSON tabanlı dilin resmi spesifikasyonu
Workflow Durumları Referansı (yeni sekmede açılır) - Tüm durum türleri için dokümantasyon: Task, Choice, Parallel, Map, Wait, Pass, Succeed, Fail
AWS Step Functions Servis Entegrasyonları (yeni sekmede açılır) - Desteklenen AWS servis entegrasyonları ve üç SDK entegrasyon deseni: istek-yanıt, sync ve callback
AWS Step Functions Fiyatlandırması (yeni sekmede açılır) - Örnek maliyet hesaplamalarıyla Standard ve Express Workflow’lar için durum geçişi fiyatlandırma modeli

İlgili yazılar

AppSync Subscription'larını AppSync Dışından Tetiklemek

AppSync subscription'ları yalnızca mutation ile tetiklenir. Downstream BFF olaylarını NONE veri kaynaklı bir mutation'a EventBridge ve CDK ile köprülemeyi inceliyorum.

aws · graphql · serverless +4

19 Haziran 2026

AWS Bedrock AgentCore'u CDK ile deploy etmek: hızlı başlangıç

AgentCore Runtime üzerinde minimal bir Strands agent'ı CDK ile deploy etme rehberi: parametrize stack, arm64 build, deploy ve invoke, IAM ve Marketplace ön koşulları.

aws-bedrock · ai-agents · aws-cdk +3

5 Mayıs 2026

AWS AppSync & GraphQL: Production-Ready Real-time API'ler Geliştirmek

AWS AppSync ile ölçeklenebilir real-time API'ler: JavaScript resolver'lar, subscription filtering, caching stratejileri ve infrastructure as code pattern'leri.

aws · graphql · serverless +4

14 Aralık 2025

SNS/SQS Cross-Account Fan-Out: AWS'de Multi-Account Event Dağıtımı

Amazon SNS ve SQS ile güvenli cross-account event dağıtımı: IAM policy'leri, KMS şifreleme, AWS CDK kurulumu ve production'da karşılaşılan yaygın sorunlar.

aws · sns · sqs +6

10 Aralık 2025

Saga Pattern ile Dağıtık Transaction'lar: ACID Olmadan Consistency Sağlamak

Microservices mimarisinde AWS Step Functions ve EventBridge kullanarak Saga pattern implementasyonu: idempotency, compensation logic ve production-ready pattern'ler.

design-patterns · distributed-systems · microservices +4

7 Aralık 2025

Orchestration Zorluğu#

Workflow Türleri#

Standard Workflow’lar#

Express Workflow’lar#

Maliyet Karşılaştırması#

Amazon States Language Pattern’leri#

Data Flow Kontrolü#

ResultPath ve Data Transformation#

Context Object Variable’ları#

Error Handling ve Retry Stratejileri#

Retry ile Exponential Backoff#

Error Catching ve Compensation#

Lambda Error Handling#

Circuit Breaker Pattern#

Büyük Ölçekli Processing için Distributed Map#

Temel Distributed Map#

CSV ve JSONL Processing#

Ölçekte Concurrency ve Maliyet#

Task Token’larla Callback Pattern#

Human Approval Workflow#

Approval Request’leri İşleme#

Approval Decision Callback#

Direct Service Integration’lar#

DynamoDB Integration#

SNS ve SQS Integration#

ECS Task’ı .sync ile Çalıştırma#

SDK Service Integration#

Production Implementation Pattern#

Maliyet Optimizasyon Stratejileri#

State Transition Sayısını Azaltma#

Batch Processing#

Direct Service Integration’lar#

Monitoring ve Observability#

CloudWatch Metric’leri#

CloudWatch Dashboard#

X-Ray Tracing#

Alternatif Araçlar#

Temporal#

AWS MWAA (Managed Airflow)#

EventBridge Pipes#

Sık Yapılan Hatalar#

Varsayılanı Seçmek#

Kaynaklar#

İlgili yazılar