Data Transforms in Kubernetes Quickstart

package main

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

func main() {
	// Make sure to register your callback and perform other setup in main
	transform.OnRecordWritten(copyRecordsToOutput)
}

// This will be called for each record in the input topic.
// The records returned will be written to the output topic.
func copyRecordsToOutput(event transform.WriteEvent, writer transform.RecordWriter) error {
	return writer.Write(event.Record())
}

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

fn main() {
	// Make sure to register your callback and perform other setup in main
	on_record_written(copy_records_to_output);
}

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

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

// Register your callback function in the entry point of your script.
onRecordWritten(copyRecordsToOutput);

// This function will be called for each record in the input topic.
// The records returned will be written to the output topic.
function copyRecordsToOutput(event, writer) {
  writer.write(event.record);
}

package main

import (
	"bytes"
	"encoding/csv"
	"encoding/json"
	"errors"
	"io"
	"strconv"
	"github.com/redpanda-data/redpanda/src/transform-sdk/go/transform"
)

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

type ItemQuantity struct {
	Item     string `json:"item"`
	Quantity int    `json:"quantity"`
}

func csvToJsonTransform(event transform.WriteEvent, writer transform.RecordWriter) error {
	// The input data is a CSV (without a header row) that is structured as:
	// key, item, quantity
	reader := csv.NewReader(bytes.NewReader(event.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
		quantity, err := strconv.Atoi(row[2])
		if err != nil {
			return err
		}
		// Marshall the JSON value
		iq := ItemQuantity{
			Item:     row[1],
			Quantity: quantity,
		}
		v, err := json.Marshal(&iq)
		if err != nil {
			return err
		}
		// Add the output record using the first column as the key.
		r := transform.Record{
			Key:   []byte(row[0]),
			Value: v,
		}
		if err := writer.Write(r); err != nil {
			return err
		}
	}
	return nil
}

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

onRecordWritten(csvToJsonTransform);

function csvToJsonTransform(event, writer) {
  // The input data is a CSV (without a header row) that is structured as:
  // key, item, quantity
  const input = event.record.value.text();
  const rows = input.split('\n');

  for (const row of rows) {
    const columns = row.split(',');

    if (columns.length !== 2) {
      throw new Error('unexpected number of columns');
    }

    const quantity = parseInt(columns[1], 10);
    if (isNaN(quantity)) {
      throw new Error('invalid quantity');
    }

    const itemQuantity = {
      item: columns[0],
      quantity: quantity,
    };
    event.record.value = JSON.stringify(itemQuantity);
    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(())
}

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

onRecordWritten(filterValidJson);

function filterValidJson(event, writer) {
  const recordValue = event.record.value.text();

  if (isValidJson(recordValue)) {
    writer.write(event.record);
  } else {
    console.log('Invalid JSON detected')
  }
}

function isValidJson(str) {
  try {
    JSON.parse(str);
    return true;
  } catch (e) {
    return false;
  }
}