diff --git a/compress/benchmark_test.go b/compress/benchmark_test.go
new file mode 100644
index 00000000..42c36ba8
--- /dev/null
+++ b/compress/benchmark_test.go
@@ -0,0 +1,17 @@
+package compress
+
+import (
+	"testing"
+)
+
+// BenchmarkNew-24    	   55165	     22851 ns/op	   23884 B/op	       2 allocs/op
+func BenchmarkNew(b *testing.B) {
+	b.ReportAllocs()
+	c, _ := New(CompressionAlgoZstd, CompressionLevelZstdBest)
+	defer func() { _ = c.Close() }()
+
+	for i := 0; i < b.N; i++ {
+		r, _ := c.Compress(loremIpsumDolor)
+		_, _ = c.Decompress(r)
+	}
+}
diff --git a/compress/compress.go b/compress/compress.go
new file mode 100644
index 00000000..a64a15aa
--- /dev/null
+++ b/compress/compress.go
@@ -0,0 +1,154 @@
+package compress
+
+import (
+	"fmt"
+
+	"github.com/klauspost/compress/zstd"
+)
+
+// CompressionAlgorithm is the interface that wraps the compression algorithm method.
+type CompressionAlgorithm int
+
+func (c CompressionAlgorithm) String() string {
+	switch c {
+	case CompressionAlgoZstd:
+		return "zstd"
+	default:
+		return ""
+	}
+}
+
+func (c CompressionAlgorithm) isValid() bool {
+	return c == CompressionAlgoZstd
+}
+
+func NewCompressionAlgorithm(s string) (CompressionAlgorithm, error) {
+	switch s {
+	case "zstd":
+		return CompressionAlgoZstd, nil
+	default:
+		return 0, fmt.Errorf("unknown compression algorithm: %s", s)
+	}
+}
+
+// CompressionLevel is the interface that wraps the compression level method.
+type CompressionLevel int
+
+func (c CompressionLevel) String() string {
+	switch c {
+	case CompressionLevelZstdFastest:
+		return "fastest"
+	case CompressionLevelZstdDefault:
+		return "default"
+	case CompressionLevelZstdBetter:
+		return "better"
+	case CompressionLevelZstdBest:
+		return "best"
+	default:
+		return ""
+	}
+}
+
+func (c CompressionLevel) isValid() bool {
+	switch c {
+	case CompressionLevelZstdFastest,
+		CompressionLevelZstdDefault,
+		CompressionLevelZstdBetter,
+		CompressionLevelZstdBest:
+		return true
+	default:
+		return false
+	}
+}
+
+func NewCompressionLevel(s string) (CompressionLevel, error) {
+	switch s {
+	case "fastest":
+		return CompressionLevelZstdFastest, nil
+	case "default":
+		return CompressionLevelZstdDefault, nil
+	case "better":
+		return CompressionLevelZstdBetter, nil
+	case "best":
+		return CompressionLevelZstdBest, nil
+	default:
+		return 0, fmt.Errorf("unknown compression level: %s", s)
+	}
+}
+
+var (
+	CompressionAlgoZstd = CompressionAlgorithm(1)
+
+	CompressionLevelZstdFastest = CompressionLevel(zstd.SpeedFastest)
+	CompressionLevelZstdDefault = CompressionLevel(zstd.SpeedDefault) // "pretty fast" compression
+	CompressionLevelZstdBetter  = CompressionLevel(zstd.SpeedBetterCompression)
+	CompressionLevelZstdBest    = CompressionLevel(zstd.SpeedBestCompression)
+)
+
+func New(algo CompressionAlgorithm, level CompressionLevel) (*Compressor, error) {
+	if !algo.isValid() {
+		return nil, fmt.Errorf("invalid compression algorithm: %d", algo)
+	}
+	if !level.isValid() {
+		return nil, fmt.Errorf("invalid compression level: %d", level)
+	}
+
+	encoder, err := zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.EncoderLevel(level)))
+	if err != nil {
+		return nil, fmt.Errorf("cannot create zstd encoder: %w", err)
+	}
+
+	decoder, err := zstd.NewReader(nil)
+	if err != nil {
+		return nil, fmt.Errorf("cannot create zstd decoder: %w", err)
+	}
+
+	return &Compressor{
+		encoder: encoder,
+		decoder: decoder,
+	}, nil
+}
+
+type Compressor struct {
+	encoder *zstd.Encoder
+	decoder *zstd.Decoder
+}
+
+func (c *Compressor) Compress(src []byte) ([]byte, error) {
+	return c.encoder.EncodeAll(src, nil), nil
+}
+
+func (c *Compressor) Decompress(src []byte) ([]byte, error) {
+	return c.decoder.DecodeAll(src, nil)
+}
+
+func (c *Compressor) Close() error {
+	c.decoder.Close()
+	return c.encoder.Close()
+}
+
+// SerializeSettings serializes the compression settings.
+func SerializeSettings(algo CompressionAlgorithm, level CompressionLevel) string {
+	return fmt.Sprintf("%d:%d", algo, level)
+}
+
+// DeserializeSettings deserializes the compression settings.
+func DeserializeSettings(s string) (CompressionAlgorithm, CompressionLevel, error) {
+	var algoInt, levelInt int
+	_, err := fmt.Sscanf(s, "%d:%d", &algoInt, &levelInt)
+	if err != nil {
+		return 0, 0, fmt.Errorf("cannot deserialize settings: %w", err)
+	}
+
+	algo := CompressionAlgorithm(algoInt)
+	if !algo.isValid() {
+		return 0, 0, fmt.Errorf("invalid compression algorithm: %d", algoInt)
+	}
+
+	level := CompressionLevel(levelInt)
+	if !level.isValid() {
+		return 0, 0, fmt.Errorf("invalid compression level: %d", levelInt)
+	}
+
+	return algo, level, nil
+}
diff --git a/compress/compress_test.go b/compress/compress_test.go
new file mode 100644
index 00000000..f18f7579
--- /dev/null
+++ b/compress/compress_test.go
@@ -0,0 +1,75 @@
+package compress
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+var loremIpsumDolor = []byte(`Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.
+Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.
+Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.
+Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.`)
+
+func TestCompress(t *testing.T) {
+	compressionLevels := []CompressionLevel{
+		CompressionLevelZstdFastest,
+		CompressionLevelZstdDefault,
+		CompressionLevelZstdBetter,
+		CompressionLevelZstdBest,
+	}
+	for _, level := range compressionLevels {
+		c, err := New(CompressionAlgoZstd, level)
+		require.NoError(t, err)
+
+		t.Cleanup(func() { _ = c.Close() })
+
+		compressed, err := c.Compress(loremIpsumDolor)
+		require.NoError(t, err)
+		require.Less(t, len(compressed), len(loremIpsumDolor))
+
+		decompressed, err := c.Decompress(compressed)
+		require.NoError(t, err)
+		require.Equal(t, string(loremIpsumDolor), string(decompressed))
+	}
+}
+
+func TestSerialization(t *testing.T) {
+	algo, err := NewCompressionAlgorithm("zstd")
+	require.NoError(t, err)
+	require.Equal(t, CompressionAlgoZstd, algo)
+
+	level, err := NewCompressionLevel("best")
+	require.NoError(t, err)
+	require.Equal(t, CompressionLevelZstdBest, level)
+
+	serialized := SerializeSettings(algo, level)
+	require.Equal(t, "1:4", serialized)
+
+	algo, level, err = DeserializeSettings(serialized)
+	require.NoError(t, err)
+	require.Equal(t, CompressionAlgoZstd, algo)
+	require.Equal(t, CompressionLevelZstdBest, level)
+}
+
+func TestDeserializationError(t *testing.T) {
+	// valid algo is 1
+	// valid level is 1-4
+	testCases := []string{
+		"0:0", "0:1", "1:0", "2:1", "1:5",
+	}
+	for _, tc := range testCases {
+		_, _, err := DeserializeSettings(tc)
+		require.Error(t, err)
+	}
+}
+
+func TestNewError(t *testing.T) {
+	c, err := New(CompressionAlgorithm(0), CompressionLevelZstdDefault)
+	require.Nil(t, c)
+	require.Error(t, err)
+
+	c, err = New(CompressionAlgoZstd, CompressionLevel(0))
+	require.Nil(t, c)
+	require.Error(t, err)
+}
diff --git a/go.mod b/go.mod
index 189d2afb..c30b08d9 100644
--- a/go.mod
+++ b/go.mod
@@ -18,6 +18,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/joho/godotenv v1.5.1
 	github.com/json-iterator/go v1.1.12
+	github.com/klauspost/compress v1.17.9
 	github.com/lib/pq v1.10.9
 	github.com/linkedin/goavro/v2 v2.13.0
 	github.com/melbahja/goph v1.4.0
@@ -61,13 +62,6 @@ require (
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1
 )
 
-require (
-	github.com/containerd/platforms v0.2.1 // indirect
-	github.com/go-ini/ini v1.67.0 // indirect
-	github.com/moby/docker-image-spec v1.3.1 // indirect
-	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-)
-
 require (
 	cloud.google.com/go v0.115.1 // indirect
 	cloud.google.com/go/auth v0.9.1 // indirect
@@ -84,6 +78,7 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/containerd/continuity v0.4.3 // indirect
+	github.com/containerd/platforms v0.2.1 // indirect
 	github.com/coreos/go-semver v0.3.0 // indirect
 	github.com/coreos/go-systemd/v22 v22.3.2 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
@@ -95,6 +90,7 @@ require (
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsouza/fake-gcs-server v1.49.3
+	github.com/go-ini/ini v1.67.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
@@ -116,16 +112,17 @@ require (
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/heetch/avro v0.4.5 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
-	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/kr/fs v0.1.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mattn/go-ieproxy v0.0.1 // indirect
 	github.com/minio/md5-simd v1.1.2 // indirect
+	github.com/moby/docker-image-spec v1.3.1 // indirect
 	github.com/moby/term v0.5.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.1.0 // indirect
 	github.com/opencontainers/runc v1.1.13 // indirect