Develop Data Transforms

package main

import (
    "log"
    "github.com/redpanda-data/redpanda/src/transform-sdk/go/transform"
)

func main() {
    transform.OnRecordWritten(myTransform)
}

func myTransform(event transform.WriteEvent, writer transform.RecordWriter) error {
  record := event.Record()
  if record.Key == nil {
    // Handle the error internally by logging it
    log.Println("Error: Record key is nil")
    // Skip this record and continue to process other records
    return nil
  }
  // Allow errors with writes to escape
  return writer.Write(record)
}

use redpanda_transform_sdk::*;
use log::error;

fn main() {
  // Set up logging
  env_logger::init();
  on_record_written(my_transform);
}

fn my_transform(event: WriteEvent, writer: &mut RecordWriter) -> anyhow::Result<()> {
  let record = event.record;
  if record.key().is_none() {
    // Handle the error internally by logging it
    error!("Error: Record key is nil");
    // Skip this record and continue to process other records
    return Ok(());
  }
  // Allow errors with writes to escape
  return writer.write(record)
}

import { onRecordWritten } from "@redpanda-data/transform-sdk";

// Register your transform function.
onRecordWritten((event, writer) => {
  const record = event.record;
  if (!record.key) {
    // Handle the error internally by logging it
    console.error("Error: Record key is nil");
    // Skip this record and continue to process other records
    return;
  }
  // Allow errors with writes to escape
  writer.write(record);
});

package main

import (
  "fmt"
  "os"
	"github.com/redpanda-data/redpanda/src/transform-sdk/go/transform"
)

func main() {
  // Check environment variables before registering the transform function.
  outputTopic1, ok := os.LookupEnv("REDPANDA_OUTPUT_TOPIC_1")
  if ok {
    fmt.Printf("Output topic 1: %s\n", outputTopic1)
  } else {
    fmt.Println("Only one output topic is set")
  }

  // Register your transform function.
  transform.OnRecordWritten(myTransform)
}

func myTransform(event transform.WriteEvent, writer transform.RecordWriter) error {
  return writer.Write(event.Record())
}

use redpanda_transform_sdk::*;
use std::env;
use log::error;

fn main() {
  // Set up logging
  env_logger::init();

  // Check environment variables before registering the transform function.
  match env::var("REDPANDA_OUTPUT_TOPIC_1") {
    Ok(output_topic_1) => println!("Output topic 1: {}", output_topic_1),
    Err(_) => println!("Only one output topic is set"),
  }

  // Register your transform function.
  on_record_written(my_transform);
}

fn my_transform(_event: WriteEvent, _writer: &mut RecordWriter) -> anyhow::Result<()> {
  Ok(())
}

import { onRecordWritten } from "@redpanda-data/transform-sdk";

// Check environment variables before registering the transform function.
const outputTopic1 = process.env.REDPANDA_OUTPUT_TOPIC_1;
if (outputTopic1) {
  console.log(`Output topic 1: ${outputTopic1}`);
} else {
  console.log("Only one output topic is set");
}

// Register your transform function.
onRecordWritten((event, writer) => {
  return writer.write(event.record);
});

import (
	"encoding/json"
	"github.com/redpanda-data/redpanda/src/transform-sdk/go/transform"
)

func main() {
	transform.OnRecordWritten(filterValidJson)
}

func filterValidJson(event transform.WriteEvent, writer transform.RecordWriter) error {
	if json.Valid(event.Record().Value) {
		return w.Write(e.Record())
	}
	// Send invalid records to separate topic
	return writer.Write(e.Record(), transform.ToTopic("invalid-json"))
}

use anyhow::Result;
use redpanda_transform_sdk::*;

fn main() {
	on_record_written(filter_valid_json);
}

fn filter_valid_json(event: WriteEvent, writer: &mut RecordWriter) -> Result<()> {
	let value = event.record.value().unwrap_or_default();
	if serde_json::from_slice::<serde_json::Value>(value).is_ok() {
		writer.write(event.record)?;
	} else {
		// Send invalid records to separate topic
		writer.write_with_options(event.record, WriteOptions::to_topic("invalid-json"))?;
	}
	Ok(())
}