Run Data Transforms in Kubernetes

package main

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

// This example shows the basic usage of the package:
// This transform does nothing but copy the same data from an
// input topic to an output topic.
func main() {
	// Make sure to register your callback and perform other setup in main
	transform.OnRecordWritten(identityTransform)
}

// This will be called for each record in the source topic.
//
// The output records returned will be written to the destination topic.
func identityTransform(e transform.WriteEvent, w transform.RecordWriter) error {
	return w.Write(e.Record())
}

package main

import (
	"bytes"
	"encoding/csv"
	"encoding/json"
	"errors"
	"io"
	"strconv"

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

// This example shows a transform that converts CSV inputs into JSON outputs.
func main() {
	transform.OnRecordWritten(csvToJsonTransform)
}

type Foo struct {
	A string `json:"a"`
	B int    `json:"b"`
}

func csvToJsonTransform(e transform.WriteEvent, w transform.RecordWriter) error {
	// The input data is a CSV (without a header row) that is the structure of:
	// key, a, b
	reader := csv.NewReader(bytes.NewReader(e.Record().Value))
	// Improve performance by reusing the result slice.
	reader.ReuseRecord = true
	for {
		row, err := reader.Read()
		if err == io.EOF {
			break
		} else if err != nil {
			return err
		}
		if len(row) != 3 {
			return errors.New("unexpected number of rows")
		}
		// Convert the last column into an int
		b, err := strconv.Atoi(row[2])
		if err != nil {
			return err
		}
		// Marshal our JSON value
		f := Foo{
			A: row[1],
			B: b,
		}
		v, err := json.Marshal(&f)
		if err != nil {
			return err
		}
		// Add our output record using the first column as the key.
		r := transform.Record{
			Key:   []byte(row[0]),
			Value: v,
		}
		if err := w.Write(r); err != nil {
			return err
		}
	}
	return nil
}

import (
	"encoding/json"

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

// This example shows a filter that outputs only valid JSON into the
// output topic.
func main() {
	transform.OnRecordWritten(filterValidJson)
}

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

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

// This example shows the basic usage of the crate:
// This transform does nothing but copy the same data from an
// input topic to an output topic.
fn main() {
    // Make sure to register your callback and perform other setup in main
    on_record_written(my_transform);
}

// This will be called for each record in the source topic.
//
// The output records returned will be written to the destination topic.
fn my_transform(event: WriteEvent, writer: &mut RecordWriter) -> Result<()> {
    writer.write(event.record)?;
    Ok(())
}

use anyhow::Result;
use redpanda_transform_sdk::*;
use serde::{Deserialize, Serialize};

// This example shows a transform that converts CSV inputs into JSON outputs.
fn main() {
    on_record_written(my_transform);
}

#[derive(Serialize, Deserialize)]
struct Foo {
    a: String,
    b: i32,
}

fn my_transform(event: WriteEvent, writer: &mut RecordWriter) -> Result<()> {
    // The input data is a CSV (without a header row) that is defined as our Foo structure.
    let mut reader = csv::Reader::from_reader(event.record.value().unwrap_or_default());
    // For each record in our CSV
    for result in reader.deserialize() {
        let foo: Foo = result?;
        // Convert it to JSON
        let value = serde_json::to_vec(&foo)?;
        // Then output it with the same key.
        writer.write(BorrowedRecord::new(event.record.key(), Some(&value)))?;
    }
    Ok(())
}

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

// This example shows a filter that outputs only valid JSON to the output topic.
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
        write.write_with_options(event.record, WriteOptions::to_topic("invalid-json"))?;
    }
    Ok(())
}