Data Transforms Sandbox

curl -SLO https://dl.redpanda.com/i5iqIdFA2acjP1CQ/redpanda-wasm-transforms-tech-preview/raw/versions/0.0.0-20230830git604fcce/rpk-darwin-arm64.tar.gz

curl -SLO https://dl.redpanda.com/i5iqIdFA2acjP1CQ/redpanda-wasm-transforms-tech-preview/raw/versions/0.0.0-20230830git604fcce/rpk-darwin-amd64.tar.gz

curl -SLO https://dl.redpanda.com/i5iqIdFA2acjP1CQ/redpanda-wasm-transforms-tech-preview/raw/versions/0.0.0-20230830git604fcce/rpk-linux-amd64.tar.gz

curl -SLO https://dl.redpanda.com/i5iqIdFA2acjP1CQ/redpanda-wasm-transforms-tech-preview/raw/versions/0.0.0-20230830git604fcce/rpk-linux-arm64.tar.gz

package main

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

// 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
	redpanda.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 redpanda.WriteEvent) ([]redpanda.Record, error) {
	return []redpanda.Record{e.Record()}, nil
}

package main

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

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

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

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

func csvToJsonTransform(e redpanda.WriteEvent) ([]redpanda.Record, 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
	output := []redpanda.Record{}
	for {
		row, err := reader.Read()
		if err == io.EOF {
			break
		} else if err != nil {
			return nil, err
		}
		if len(row) != 3 {
			return nil, errors.New("unexpected number of rows")
		}
		// Convert the last column into an int
		b, err := strconv.Atoi(row[2])
		if err != nil {
			return nil, err
		}
		// Marshal our JSON value
		f := Foo{
			A: row[1],
			B: b,
		}
		v, err := json.Marshal(&f)
		if err != nil {
			return nil, err
		}
		// Add our output record using the first column as the key.
		output = append(output, redpanda.Record{
			Key:   []byte(row[0]),
			Value: v,
		})

	}
	return output, nil
}

import (
	"encoding/json"

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

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

func filterValidJson(e redpanda.WriteEvent) ([]redpanda.Record, error) {
	v := []redpanda.Record{}
	if json.Valid(e.Record().Value) {
		v = append(v, e.Record())
	}
	return v, nil
}