diff --git a/cmd/graphplan/main.go b/cmd/graphplan/main.go
index 3b8843e..56b93f8 100644
--- a/cmd/graphplan/main.go
+++ b/cmd/graphplan/main.go
@@ -1,16 +1,18 @@
 package graphplan
 
 import (
-	"log"
 	"path/filepath"
 
 	"github.com/bmeg/sifter/graphplan"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/playbook"
 	"github.com/spf13/cobra"
 )
 
 var outScriptDir = ""
 var outDataDir = "./"
+var objectExclude = []string{}
+var verbose bool = false
 
 // Cmd is the declaration of the command line
 var Cmd = &cobra.Command{
@@ -19,6 +21,10 @@ var Cmd = &cobra.Command{
 	Args:  cobra.MinimumNArgs(1),
 	RunE: func(cmd *cobra.Command, args []string) error {
 
+		if verbose {
+			logger.Init(true, false)
+		}
+
 		scriptPath, _ := filepath.Abs(args[0])
 
 		/*
@@ -40,10 +46,10 @@ var Cmd = &cobra.Command{
 		if sifterErr := playbook.ParseFile(scriptPath, &pb); sifterErr == nil {
 			if len(pb.Pipelines) > 0 || len(pb.Inputs) > 0 {
 				err := graphplan.NewGraphBuild(
-					&pb, outScriptDir, outDataDir,
+					&pb, outScriptDir, outDataDir, objectExclude,
 				)
 				if err != nil {
-					log.Printf("Error: %s\n", err)
+					logger.Error("Parse Error", "error", err)
 				}
 			}
 		}
@@ -54,6 +60,8 @@ var Cmd = &cobra.Command{
 
 func init() {
 	flags := Cmd.Flags()
+	flags.BoolVarP(&verbose, "verbose", "v", verbose, "Verbose logging")
 	flags.StringVarP(&outScriptDir, "dir", "C", outScriptDir, "Change Directory for script base")
 	flags.StringVarP(&outDataDir, "out", "o", outDataDir, "Change output Directory")
+	flags.StringArrayVarP(&objectExclude, "exclude", "x", objectExclude, "Object Exclude")
 }
diff --git a/cmd/inspect/main.go b/cmd/inspect/main.go
index e01909c..4375f45 100644
--- a/cmd/inspect/main.go
+++ b/cmd/inspect/main.go
@@ -3,16 +3,15 @@ package inspect
 import (
 	"encoding/json"
 	"fmt"
-	"log"
 	"path/filepath"
 
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/playbook"
 	"github.com/bmeg/sifter/task"
 	"github.com/spf13/cobra"
 )
 
 var outDir string = ""
-var inputFile string = ""
 var cmdInputs map[string]string
 
 // Cmd is the declaration of the command line
@@ -31,12 +30,12 @@ var Cmd = &cobra.Command{
 
 		pb := playbook.Playbook{}
 		if err := playbook.ParseFile(playFile, &pb); err != nil {
-			log.Printf("%s", err)
+			logger.Info("%s", err)
 			return err
 		}
 		var err error
 		inputs, err = pb.PrepConfig(inputs, "./")
-		log.Printf("inputs: %s", inputs)
+		logger.Info("inputs: %s", inputs)
 		if err != nil {
 			return err
 		}
@@ -45,7 +44,7 @@ var Cmd = &cobra.Command{
 			outDir = pb.GetDefaultOutDir()
 		}
 
-		log.Printf("outdir: %s", outDir)
+		logger.Info("outdir: %s", outDir)
 
 		p, _ := filepath.Abs(playFile)
 		baseDir := filepath.Dir(p)
diff --git a/cmd/run/main.go b/cmd/run/main.go
index a379679..5163223 100644
--- a/cmd/run/main.go
+++ b/cmd/run/main.go
@@ -1,37 +1,35 @@
 package run
 
 import (
-	"io/ioutil"
-	"log"
+	"io"
 	"os"
 
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/playbook"
 
 	"github.com/spf13/cobra"
 )
 
-var workDir string = "./"
 var outDir string = ""
-var resume string = ""
-var graph string = ""
 var inputFile string = ""
-var toStdout bool
-var keep bool
+var verbose bool = false
 var cmdInputs map[string]string
 
-var proxy = ""
-var port = 8888
-
 // Cmd is the declaration of the command line
 var Cmd = &cobra.Command{
 	Use:   "run <script>",
 	Short: "Run sifter script",
 	Args:  cobra.MinimumNArgs(1),
 	RunE: func(cmd *cobra.Command, args []string) error {
+
+		if verbose {
+			logger.Init(true, false)
+		}
+
 		inputs := map[string]string{}
 		if inputFile != "" {
 			if err := playbook.ParseStringFile(inputFile, &inputs); err != nil {
-				log.Printf("%s", err)
+				logger.Error("%s", err)
 				return err
 			}
 		}
@@ -40,9 +38,9 @@ var Cmd = &cobra.Command{
 		}
 		for _, playFile := range args {
 			if playFile == "-" {
-				yaml, err := ioutil.ReadAll(os.Stdin)
+				yaml, err := io.ReadAll(os.Stdin)
 				if err != nil {
-					log.Printf("%s", err)
+					logger.Error("%s", err)
 					return err
 				}
 				pb := playbook.Playbook{}
@@ -63,15 +61,7 @@ var Cmd = &cobra.Command{
 
 func init() {
 	flags := Cmd.Flags()
-	flags.StringVarP(&workDir, "workdir", "w", workDir, "Workdir")
-	flags.BoolVarP(&toStdout, "stdout", "s", toStdout, "To STDOUT")
-	flags.BoolVarP(&keep, "keep", "k", keep, "Keep Working Directory")
-	flags.StringVarP(&outDir, "out", "o", outDir, "Output Dir")
-	flags.StringVarP(&resume, "resume", "r", resume, "Resume Directory")
-	flags.StringVarP(&graph, "graph", "g", graph, "Output to graph")
-
-	flags.StringVar(&proxy, "proxy", proxy, "Proxy site")
-	flags.IntVar(&port, "port", port, "Proxy Port")
+	flags.BoolVarP(&verbose, "verbose", "v", verbose, "Verbose logging")
 	flags.StringToStringVarP(&cmdInputs, "inputs", "i", cmdInputs, "Input variables")
 	flags.StringVarP(&inputFile, "inputfile", "f", inputFile, "Input variables file")
 }
diff --git a/cmd/run/run.go b/cmd/run/run.go
index 07ad0b9..a5607a6 100644
--- a/cmd/run/run.go
+++ b/cmd/run/run.go
@@ -1,27 +1,24 @@
 package run
 
 import (
-	"fmt"
-	"log"
 	"os"
 	"path/filepath"
 
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/playbook"
 	"github.com/bmeg/sifter/task"
 )
 
 func ExecuteFile(playFile string, workDir string, outDir string, inputs map[string]string) error {
-	log.Printf("Starting: %s\n", playFile)
-	log.Default().SetPrefix(fmt.Sprintf("%s: ", playFile))
+	logger.Info("Starting", "playFile", playFile)
 	pb := playbook.Playbook{}
 	if err := playbook.ParseFile(playFile, &pb); err != nil {
-		log.Printf("%s", err)
+		logger.Error("%s", err)
 		return err
 	}
 	a, _ := filepath.Abs(playFile)
 	baseDir := filepath.Dir(a)
-	log.Printf("basedir: %s", baseDir)
-	log.Printf("playbook: %#v", pb)
+	logger.Debug("parsed file", "baseDir", baseDir, "playbook", pb)
 	return Execute(pb, baseDir, workDir, outDir, inputs)
 }
 
@@ -39,7 +36,7 @@ func Execute(pb playbook.Playbook, baseDir string, workDir string, outDir string
 	if err != nil {
 		return err
 	}
-	log.Printf("Outdir: %s", outDir)
+	logger.Debug("Running", "outDir", outDir)
 
 	t := task.NewTask(pb.Name, baseDir, workDir, outDir, nInputs)
 	err = pb.Execute(t)
diff --git a/evaluate/docker_python.go b/evaluate/docker_python.go
index 97a9b04..7ffdc25 100644
--- a/evaluate/docker_python.go
+++ b/evaluate/docker_python.go
@@ -1,7 +1,6 @@
 package evaluate
 
 import (
-	"log"
 	"os"
 	"time"
 
@@ -10,6 +9,7 @@ import (
 	"os/exec"
 	"strings"
 
+	"github.com/bmeg/sifter/logger"
 	grpc "google.golang.org/grpc"
 )
 
@@ -49,7 +49,7 @@ func StartDockerExecutor(dockerImage string) (Runner, error) {
 		return nil, err
 	}
 	id := strings.Trim(string(out), " \r\n\t")
-	log.Printf("Started Container: %s", id)
+	logger.Debug("Started Container: %s", id)
 	cmd = exec.Command("docker", "port", id)
 	out, err = cmd.Output()
 	if err != nil {
@@ -62,7 +62,7 @@ func StartDockerExecutor(dockerImage string) (Runner, error) {
 	var conn *grpc.ClientConn
 
 	for i := 0; i < 5; i++ {
-		log.Printf("Contacting: %s", serverAddr)
+		logger.Debug("Contacting: %s", serverAddr)
 		conn, err = grpc.Dial(serverAddr, grpc.WithInsecure(), grpc.WithBlock())
 		if err == nil {
 			break
@@ -70,7 +70,7 @@ func StartDockerExecutor(dockerImage string) (Runner, error) {
 		time.Sleep(1 * time.Second)
 	}
 	if err != nil {
-		log.Printf("Returning err: %s", err)
+		logger.Error("Returning err: %s", err)
 		return nil, err
 	}
 	client := NewExecutorClient(conn)
@@ -86,7 +86,7 @@ func (run *DockerRunner) Call(in *Input) (*Result, error) {
 }
 
 func (run *DockerRunner) Close() {
-	log.Printf("Closing docker %s", run.containerID)
+	logger.Debug("Closing docker %s", run.containerID)
 	run.conn.Close()
 	cmd := exec.Command("docker", "kill", run.containerID)
 	cmd.Run()
diff --git a/evaluate/gpython.go b/evaluate/gpython.go
index 558c6e7..420897f 100644
--- a/evaluate/gpython.go
+++ b/evaluate/gpython.go
@@ -2,8 +2,8 @@ package evaluate
 
 import (
 	"fmt"
-	"log"
 
+	"github.com/bmeg/sifter/logger"
 	"github.com/go-python/gpython/py"
 	_ "github.com/go-python/gpython/stdlib" // Load modiles in gpython environment
 )
@@ -80,7 +80,7 @@ func PyObject(i interface{}) py.Object {
 	} else if i == nil {
 		return py.None
 	} else {
-		log.Printf("gpython conversion Not found: %T", i)
+		logger.Error("gpython conversion Not found: %T", i)
 	}
 	return nil
 }
@@ -127,7 +127,7 @@ func FromPyObject(i py.Object) interface{} {
 	} else if i == py.None {
 		return nil
 	} else {
-		log.Printf("gpython conversion Not found: %T", i)
+		logger.Error("gpython conversion Not found: %T", i)
 	}
 	return nil
 }
@@ -138,7 +138,7 @@ type PyCode struct {
 
 func PyCompile(codeStr string) (*PyCode, error) {
 
-	log.Printf("Gpython compile: %s", codeStr)
+	logger.Debug("Gpython compile: %s", codeStr)
 
 	opts := py.ContextOpts{SysArgs: []string{}, SysPaths: []string{}}
 	ctx := py.NewContext(opts)
@@ -166,9 +166,9 @@ func (p *PyCode) Evaluate(method string, inputs ...map[string]interface{}) (map[
 	out, err := py.Call(fun, in, nil)
 	if err != nil {
 		py.TracebackDump(err)
-		log.Printf("Error Inputs: %#v", inputs)
-		log.Printf("Error Inputs: %#v", in)
-		log.Printf("Map Error: %s", err)
+		logger.Error("Error Inputs: %#v", inputs)
+		logger.Error("Error Inputs: %#v", in)
+		logger.Error("Map Error: %s", err)
 		return nil, err
 	}
 	o := FromPyObject(out)
@@ -188,9 +188,9 @@ func (p *PyCode) EvaluateArray(method string, inputs ...map[string]interface{})
 	out, err := py.Call(fun, in, nil)
 	if err != nil {
 		py.TracebackDump(err)
-		log.Printf("Error Inputs: %#v", inputs)
-		log.Printf("Error Inputs: %#v", in)
-		log.Printf("Map Error: %s", err)
+		logger.Error("Error Inputs: %#v", inputs)
+		logger.Error("Error Inputs: %#v", in)
+		logger.Error("Map Error: %s", err)
 		return nil, err
 	}
 	o := FromPyObject(out)
@@ -210,7 +210,7 @@ func (p *PyCode) EvaluateBool(method string, inputs ...map[string]interface{}) (
 	out, err := py.Call(fun, in, nil)
 	if err != nil {
 		py.TracebackDump(err)
-		log.Printf("Map Error: %s", err)
+		logger.Error("Map Error: %s", err)
 		return false, err
 	}
 	o := FromPyObject(out)
diff --git a/evaluate/python.go b/evaluate/python.go
index 4a2787e..c14286d 100644
--- a/evaluate/python.go
+++ b/evaluate/python.go
@@ -9,12 +9,12 @@ import (
 	//"bytes"
 	"context"
 	"encoding/json"
-	"io/ioutil"
 	"os/exec"
 	"path/filepath"
 	"strconv"
 	"strings"
 
+	"github.com/bmeg/sifter/logger"
 	grpc "google.golang.org/grpc"
 )
 
@@ -66,7 +66,7 @@ func (d PythonProcessor) EvaluateArray(inputs ...map[string]interface{}) ([]any,
 func (d PythonProcessor) EvaluateBool(inputs ...map[string]interface{}) (bool, error) {
 	i, err := json.Marshal(inputs)
 	if err != nil {
-		log.Printf("Serialization Error: %s", err)
+		logger.Error("Serialization Error: %s", err)
 		return false, err
 	}
 	out, err := d.runner.Call(&Input{Data: string(i), Code: d.fNum})
@@ -106,7 +106,7 @@ type LocalRunner struct {
 
 func StartLocalExecutor(workdir string) (Runner, error) {
 
-	d, err := ioutil.TempDir(workdir, "sifterexec_") //TODO: use directory from user config
+	d, err := os.MkdirTemp(workdir, "sifterexec_") //TODO: use directory from user config
 	if err != nil {
 		return nil, err
 	}
@@ -133,7 +133,7 @@ func StartLocalExecutor(workdir string) (Runner, error) {
 	stdout, _ := cmd.StdoutPipe()
 	err = cmd.Start()
 	if err != nil {
-		log.Printf("Error starting python: %s", err)
+		logger.Error("Error starting python: %s", err)
 		return nil, err
 	}
 
@@ -152,7 +152,7 @@ func StartLocalExecutor(workdir string) (Runner, error) {
 					d := buf[:n]
 					out = append(out, d...)
 				}
-				log.Printf("Read %d (%s) Buffer: %s", n, ierr, string(out))
+				logger.Debug("Read %d (%s) Buffer: %s", n, ierr, string(out))
 				if strings.Contains(string(out), "\n") {
 					t := strings.Split(string(out), "\n")
 					log.Printf("Return port: %s", out)
@@ -164,9 +164,9 @@ func StartLocalExecutor(workdir string) (Runner, error) {
 			}
 			if ierr != nil {
 				if ierr == io.EOF {
-					log.Printf("Executor closed")
+					logger.Debug("Executor closed")
 					if cmd.ProcessState.ExitCode() != 0 {
-						log.Printf("Executor error: %d", cmd.ProcessState.ExitCode())
+						logger.Error("Executor error: %d", cmd.ProcessState.ExitCode())
 					}
 					ierr = nil
 				}
diff --git a/extractors/avro_load.go b/extractors/avro_load.go
index b81034d..dc2975d 100644
--- a/extractors/avro_load.go
+++ b/extractors/avro_load.go
@@ -2,11 +2,11 @@ package extractors
 
 import (
 	"fmt"
-	"log"
 	"os"
 
 	"github.com/bmeg/sifter/config"
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 
 	"github.com/linkedin/goavro/v2"
@@ -17,7 +17,7 @@ type AvroLoadStep struct {
 }
 
 func (ml *AvroLoadStep) Start(task task.RuntimeTask) (chan map[string]interface{}, error) {
-	log.Printf("Starting Avro Load")
+	logger.Debug("Starting Avro Load")
 
 	input, err := evaluate.ExpressionString(ml.Input, task.GetConfig(), nil)
 	if err != nil {
@@ -27,7 +27,7 @@ func (ml *AvroLoadStep) Start(task task.RuntimeTask) (chan map[string]interface{
 	if _, err := os.Stat(input); os.IsNotExist(err) {
 		return nil, fmt.Errorf("file not found: %s", input)
 	}
-	log.Printf("Loading: %s", input)
+	logger.Debug("Loading: %s", input)
 
 	fh, err := os.Open(input)
 	if err != nil {
@@ -51,7 +51,7 @@ func (ml *AvroLoadStep) Start(task task.RuntimeTask) (chan map[string]interface{
 		}
 		close(procChan)
 		fh.Close()
-		log.Printf("Done Loading")
+		logger.Debug("Done Loading")
 	}()
 
 	return procChan, nil
diff --git a/extractors/glob_load.go b/extractors/glob_load.go
index 22c97f6..8c57c01 100644
--- a/extractors/glob_load.go
+++ b/extractors/glob_load.go
@@ -2,12 +2,12 @@ package extractors
 
 import (
 	"fmt"
-	"log"
 	"path/filepath"
 	"sync"
 
 	"github.com/bmeg/sifter/config"
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -33,7 +33,7 @@ func (gl *GlobLoadStep) Start(task task.RuntimeTask) (chan map[string]interface{
 		return nil, err
 	}
 
-	log.Printf("Starting Glob Load: %s", input)
+	logger.Debug("Starting Glob Load", "input", input)
 	if gl.XMLLoad != nil || gl.JSONLoad != nil || gl.TableLoad != nil {
 		flist, err := filepath.Glob(input)
 		if err != nil {
@@ -48,7 +48,7 @@ func (gl *GlobLoadStep) Start(task task.RuntimeTask) (chan map[string]interface{
 		go func() {
 			defer close(sources)
 			for count, f := range flist {
-				log.Printf("Glob %d of %d", count, len(flist))
+				logger.Debug("Glob", "file", f, "num", count, "count", len(flist))
 				//var a func()
 				var a Source
 				if gl.XMLLoad != nil {
diff --git a/extractors/graph_file_reader.go b/extractors/graph_file_reader.go
index 43447cb..b5f977d 100644
--- a/extractors/graph_file_reader.go
+++ b/extractors/graph_file_reader.go
@@ -8,6 +8,7 @@ import (
 
 	"github.com/bmeg/golib"
 	"github.com/bmeg/grip/gripql"
+	"github.com/bmeg/sifter/logger"
 	"google.golang.org/protobuf/encoding/protojson"
 )
 
@@ -15,7 +16,7 @@ func LoadVertexFile(path string) (chan *gripql.Vertex, error) {
 	if _, err := os.Stat(path); os.IsNotExist(err) {
 		return nil, fmt.Errorf("file not found: %s", path)
 	}
-	log.Printf("Loading: %s", path)
+	logger.Debug("Loading: %s", path)
 
 	var reader chan []byte
 	var err error
diff --git a/extractors/gripper_load.go b/extractors/gripper_load.go
index b0fa482..36cd588 100644
--- a/extractors/gripper_load.go
+++ b/extractors/gripper_load.go
@@ -3,11 +3,11 @@ package extractors
 import (
 	"context"
 	"io"
-	"log"
 
 	"github.com/bmeg/grip/gripper"
 	"github.com/bmeg/grip/util/rpc"
 
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -17,13 +17,13 @@ type GripperLoadStep struct {
 }
 
 func (ml *GripperLoadStep) Start(task.RuntimeTask) (chan map[string]interface{}, error) {
-	log.Printf("Starting GRIPPER Load")
+	logger.Debug("Starting GRIPPER Load")
 
 	rpcConf := rpc.ConfigWithDefaults(ml.Host)
-	log.Printf("Connecting to %s", ml.Host)
+	logger.Debug("Connecting to %s", ml.Host)
 	conn, err := rpc.Dial(context.Background(), rpcConf)
 	if err != nil {
-		log.Printf("RPC Connection error: %s", err)
+		logger.Error("RPC Connection error: %s", err)
 		return nil, err
 	}
 	client := gripper.NewGRIPSourceClient(conn)
@@ -31,7 +31,7 @@ func (ml *GripperLoadStep) Start(task.RuntimeTask) (chan map[string]interface{},
 	procChan := make(chan map[string]interface{}, 100)
 	go func() {
 		req := gripper.Collection{Name: ml.Collection}
-		log.Printf("Loading: '%s'", ml.Collection)
+		logger.Debug("Loading: '%s'", ml.Collection)
 		cl, err := client.GetRows(context.Background(), &req)
 		if err == nil {
 			for {
@@ -44,9 +44,9 @@ func (ml *GripperLoadStep) Start(task.RuntimeTask) (chan map[string]interface{},
 				}
 			}
 		} else {
-			log.Printf("Error Getting rows: %s", err)
+			logger.Error("Error Getting rows: %s", err)
 		}
-		log.Printf("Done Loading")
+		logger.Debug("Done Loading")
 		close(procChan)
 	}()
 
diff --git a/extractors/json_load.go b/extractors/json_load.go
index bcfc548..6f4ccb6 100644
--- a/extractors/json_load.go
+++ b/extractors/json_load.go
@@ -3,13 +3,13 @@ package extractors
 import (
 	"encoding/json"
 	"fmt"
-	"log"
 	"os"
 	"strings"
 
 	"github.com/bmeg/golib"
 	"github.com/bmeg/sifter/config"
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 	"github.com/bmeg/sifter/transform"
 )
@@ -21,7 +21,7 @@ type JSONLoadStep struct {
 }
 
 func (ml *JSONLoadStep) Start(task task.RuntimeTask) (chan map[string]interface{}, error) {
-	log.Printf("Starting JSON Load")
+	logger.Debug("Starting JSON Load")
 	input, err := evaluate.ExpressionString(ml.Input, task.GetConfig(), nil)
 	if err != nil {
 		return nil, err
@@ -31,7 +31,7 @@ func (ml *JSONLoadStep) Start(task task.RuntimeTask) (chan map[string]interface{
 	if _, err := os.Stat(inputPath); os.IsNotExist(err) {
 		return nil, fmt.Errorf("file not found: %s", input)
 	}
-	log.Printf("Loading: %s", inputPath)
+	logger.Debug("Loading: %s", inputPath)
 
 	var reader chan []byte
 	if ml.Multiline {
diff --git a/extractors/plugin_load.go b/extractors/plugin_load.go
index 1ecbe7e..74893cb 100644
--- a/extractors/plugin_load.go
+++ b/extractors/plugin_load.go
@@ -4,13 +4,13 @@ import (
 	"bufio"
 	"encoding/json"
 	"io"
-	"log"
 	"os"
 	"os/exec"
 	"sync"
 
 	"github.com/bmeg/sifter/config"
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 	"github.com/google/shlex"
 )
@@ -20,7 +20,7 @@ type PluginLoadStep struct {
 }
 
 func (ml *PluginLoadStep) Start(task task.RuntimeTask) (chan map[string]interface{}, error) {
-	log.Printf("Starting JSON Load")
+	logger.Debug("Starting Plugin Loader")
 	cmdText, err := evaluate.ExpressionString(ml.CommandLine, task.GetConfig(), nil)
 	if err != nil {
 		return nil, err
@@ -37,12 +37,13 @@ func (ml *PluginLoadStep) Start(task task.RuntimeTask) (chan map[string]interfac
 		cmd.Dir = workdir
 		stdout, _ := cmd.StdoutPipe()
 		cmd.Stderr = os.Stderr
-		log.Printf("Starting: %#v", cmd)
+		logger.Debug("Starting: %#v", cmd)
 		err := cmd.Start()
 		if err != nil {
-			log.Printf("plugin exec error: %s", err)
+			logger.Error("plugin exec error", "error", err)
 		}
 
+		count := uint64(0)
 		wg := &sync.WaitGroup{}
 		wg.Add(1)
 		go func() {
@@ -54,7 +55,7 @@ func (ml *PluginLoadStep) Start(task task.RuntimeTask) (chan map[string]interfac
 			for err == nil {
 				line, isPrefix, err = reader.ReadLine()
 				if err != nil && err != io.EOF {
-					log.Printf("plugin (%s) input error: %s", ml.CommandLine, err)
+					logger.Error("plugin input error", "commandLine", ml.CommandLine, "error", err)
 				}
 				ln = append(ln, line...)
 				if !isPrefix {
@@ -63,22 +64,19 @@ func (ml *PluginLoadStep) Start(task task.RuntimeTask) (chan map[string]interfac
 						err := json.Unmarshal(ln, &row)
 						if err == nil {
 							procChan <- row
+							count++
 						} else {
-							log.Printf("plugin (%s) output error: %s", ml.CommandLine, err)
-							log.Printf("unmarshalled line: %s", ln)
+							logger.Error("plugin output error", "commandLine", ml.CommandLine, "error", err, "line", ln)
 						}
 						ln = []byte{}
 					}
 				}
 			}
-
 			wg.Done()
 		}()
-
-		log.Printf("plugin has exited: %s\n", ml.CommandLine)
 		wg.Wait()
-
 		close(procChan)
+		logger.Info("Plugin Summary", "name", task.GetName(), "rowCount", count, "commandLine", cmdLine)
 	}()
 	return procChan, nil
 }
diff --git a/extractors/tabular_load.go b/extractors/tabular_load.go
index c880495..5f8b9f4 100644
--- a/extractors/tabular_load.go
+++ b/extractors/tabular_load.go
@@ -4,7 +4,6 @@ import (
 	"encoding/csv"
 	"fmt"
 	"io"
-	"log"
 	"os"
 	"strings"
 
@@ -12,6 +11,7 @@ import (
 
 	"github.com/bmeg/sifter/config"
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -49,7 +49,7 @@ func buildUniqueArray(src []string) []string {
 }
 
 func (ml *TableLoadStep) Start(task task.RuntimeTask) (chan map[string]interface{}, error) {
-	log.Printf("Starting Table Load")
+	logger.Info("Starting Table Load")
 	input, err := evaluate.ExpressionString(ml.Input, task.GetConfig(), nil)
 	if err != nil {
 		return nil, err
@@ -62,7 +62,7 @@ func (ml *TableLoadStep) Start(task task.RuntimeTask) (chan map[string]interface
 	} else if s.IsDir() {
 		return nil, fmt.Errorf("input not a file: %s", inputPath)
 	}
-	log.Printf("Loading table: %s", inputPath)
+	logger.Info("Loading table", "path", inputPath)
 
 	var inputStream io.ReadCloser
 	if gfile, err := os.Open(inputPath); err == nil {
@@ -121,9 +121,9 @@ func (ml *TableLoadStep) Start(task task.RuntimeTask) (chan map[string]interface
 							err = nil
 						}
 					} else if pe.Err == csv.ErrQuote {
-						log.Printf("quote error: %s", record)
+						logger.Error("quote error", "record", record)
 					} else if pe.Err == csv.ErrBareQuote {
-						log.Printf("bare quote error: %s", record)
+						logger.Error("bare quote error", "record", record)
 					}
 				}
 			}
@@ -150,10 +150,10 @@ func (ml *TableLoadStep) Start(task task.RuntimeTask) (chan map[string]interface
 					}
 				}
 			} else {
-				log.Printf("Error: %s", err)
+				logger.Error("Error", "error", err)
 			}
 		}
-		log.Printf("Done Loading")
+		logger.Info("Done Loading")
 		close(procChan)
 	}()
 
diff --git a/extractors/transpose_load.go b/extractors/transpose_load.go
index 41c4eeb..4d6a6ad 100644
--- a/extractors/transpose_load.go
+++ b/extractors/transpose_load.go
@@ -6,7 +6,6 @@ import (
 	"encoding/binary"
 	"encoding/csv"
 	"io"
-	"log"
 	"os"
 	"path/filepath"
 	"strings"
@@ -15,6 +14,7 @@ import (
 
 	"github.com/bmeg/sifter/config"
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 	"github.com/cockroachdb/pebble"
 )
@@ -101,12 +101,12 @@ func transposeInMem(c csvReader, out chan map[string]any) error {
 			break
 		}
 		if err != nil {
-			log.Printf("Error %s", err)
+			logger.Error("Error", "error", err)
 			break
 		}
 		matrix = append(matrix, record)
 	}
-	log.Printf("Writing Transpose")
+	logger.Info("Writing Transpose")
 	l := len(matrix[0])
 	h := len(matrix)
 	columns := make([]string, h)
@@ -142,7 +142,6 @@ func copyBytes(in []byte) []byte {
 }
 
 func (pbw *pebbleBulkWrite) Set(id []byte, val []byte) error {
-	//log.Printf("Setting %x %s", id, val)
 	pbw.curSize += len(id) + len(val)
 	if pbw.highest == nil || bytes.Compare(id, pbw.highest) > 0 {
 		pbw.highest = copyBytes(id)
@@ -152,7 +151,7 @@ func (pbw *pebbleBulkWrite) Set(id []byte, val []byte) error {
 	}
 	err := pbw.batch.Set(id, val, nil)
 	if pbw.curSize > maxWriterBuffer {
-		log.Printf("Running batch Commit")
+		logger.Info("Running batch Commit")
 		pbw.batch.Commit(nil)
 		pbw.batch.Reset()
 		pbw.curSize = 0
@@ -161,7 +160,7 @@ func (pbw *pebbleBulkWrite) Set(id []byte, val []byte) error {
 }
 
 func (pbw *pebbleBulkWrite) Close() error {
-	log.Printf("Running batch close Commit")
+	logger.Info("Running batch close Commit")
 	err := pbw.batch.Commit(nil)
 	if err != nil {
 		return err
@@ -191,7 +190,7 @@ func transposeInDB(workdir string, c csvReader, out chan map[string]any) error {
 	colCount := uint64(0)
 	for row := uint64(0); ; row++ {
 		if (row % 100) == 0 {
-			log.Printf("Row: %d", row)
+			logger.Info("Row", "row", row)
 		}
 		record, err := r.Read()
 		if err == io.EOF {
@@ -207,22 +206,21 @@ func transposeInDB(workdir string, c csvReader, out chan map[string]any) error {
 			bCol := make([]byte, 8)
 			binary.BigEndian.PutUint64(bCol, col)
 			//key := bytes.Join([][]byte{bRow, bCol}, []byte{})
-			//log.Printf("Put %x", key)
 			key := bytes.Join([][]byte{bCol, bRow}, []byte{})
 			err := pbw.Set(key, []byte(record[col]))
 			if err != nil {
-				log.Printf("Put Error: %s", err)
+				logger.Error("Put Error", "message", err)
 			}
 		}
 		rowCount = row + 1
 	}
 	if err := pbw.Close(); err != nil {
-		log.Print(err)
+		logger.Error("close error", "message", err)
 	}
 
-	log.Println(db.Metrics().String())
+	logger.Info(db.Metrics().String())
 
-	log.Printf("Col/Row counts: %d %d", colCount, rowCount)
+	logger.Info("Col/Row counts", "col", colCount, "row", rowCount)
 
 	columns := []string{}
 
@@ -238,15 +236,13 @@ func transposeInDB(workdir string, c csvReader, out chan map[string]any) error {
 			columns = append(columns, string(val))
 			c.Close()
 		} else {
-			log.Printf("Column error: %s", err)
+			logger.Error("Column error", "message", err)
 		}
 	}
 
-	//log.Printf("Columns: %#v", columns)
-
 	for col := uint64(1); col < colCount; col++ {
 		if (col % 100) == 0 {
-			log.Printf("Writing Col %d", col)
+			logger.Info("Writing Col", "id", col)
 		}
 		prefix := make([]byte, 8)
 		binary.BigEndian.PutUint64(prefix, col)
@@ -259,7 +255,6 @@ func transposeInDB(workdir string, c csvReader, out chan map[string]any) error {
 			o = append(o, string(r))
 		}
 		it.Close()
-		//log.Printf("Col width: %d %d", len(columns), len(o))
 		if len(o) == len(columns) {
 			res := make(map[string]any, len(columns))
 			for i := 0; i < len(o); i++ {
@@ -292,7 +287,7 @@ func transposeInTable(workdir string, fieldSize int, c csvReader, out chan map[s
 	writer := bufio.NewWriterSize(table, 1024*10)
 	for row := int64(0); ; row++ {
 		if (row % 100) == 0 {
-			log.Printf("Row: %d", row)
+			logger.Info("Row", "i", row)
 		}
 		record, err := r.Read()
 		if err == io.EOF {
@@ -302,7 +297,7 @@ func transposeInTable(workdir string, fieldSize int, c csvReader, out chan map[s
 			colCount = int64(len(record))
 		}
 		if colCount != int64(len(record)) {
-			log.Printf("Incorrectly sized row: %d != %d", colCount, uint64(len(record)))
+			logger.Error("Incorrectly sized row", "colCount", colCount, "recordSize", uint64(len(record)))
 		}
 		for col := int64(0); col < colCount; col++ {
 			b := []byte(record[col])
@@ -316,7 +311,7 @@ func transposeInTable(workdir string, fieldSize int, c csvReader, out chan map[s
 
 	stepSize := colCount * int64(fieldSize)
 
-	log.Printf("Col/Row counts: %d %d", colCount, rowCount)
+	logger.Info("Col/Row counts", "col", colCount, "row", rowCount)
 
 	columns := []string{}
 	for row := int64(0); row < rowCount; row++ {
@@ -326,15 +321,15 @@ func transposeInTable(workdir string, fieldSize int, c csvReader, out chan map[s
 		if err == nil {
 			columns = append(columns, string(tmp[0]))
 		} else {
-			log.Printf("Column error: %s", err)
+			logger.Error("Column error", "message", err)
 		}
 	}
 
-	log.Printf("Columns: %s\n", columns)
+	logger.Info("Columns", "columns", columns)
 
 	for col := int64(1); col < colCount; col++ {
 		if (col % 100) == 0 {
-			log.Printf("Writing Col %d", col)
+			logger.Info("Writing Col", "i", col)
 		}
 		record := map[string]any{}
 		for row := int64(0); row < rowCount; row++ {
diff --git a/extractors/xml_step.go b/extractors/xml_step.go
index 9af8c13..d7653d0 100644
--- a/extractors/xml_step.go
+++ b/extractors/xml_step.go
@@ -7,12 +7,12 @@ import (
 	"encoding/xml"
 	"fmt"
 	"io"
-	"log"
 	"os"
 	"strings"
 
 	"github.com/bmeg/sifter/config"
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 
 	xj "github.com/basgys/goxml2json"
@@ -24,17 +24,16 @@ type XMLLoadStep struct {
 }
 
 func (ml *XMLLoadStep) Start(task task.RuntimeTask) (chan map[string]any, error) {
-	//log.Printf("Starting XML Load")
 	input, err := evaluate.ExpressionString(ml.Input, task.GetConfig(), nil)
 	if err != nil {
-		log.Printf("Error: %s", err)
+		logger.Error("Error open xml", "error", err)
 		return nil, err
 	}
 
 	if _, err := os.Stat(input); os.IsNotExist(err) {
 		return nil, fmt.Errorf("file not found: %s", input)
 	}
-	log.Printf("Loading: %s", input)
+	logger.Info("Loading xml", "input", input)
 
 	fhd, err := os.Open(input)
 	if err != nil {
@@ -77,7 +76,7 @@ func (ml *XMLLoadStep) Start(task task.RuntimeTask) (chan map[string]any, error)
 					// EOF means we're done.
 					break
 				} else if err != nil {
-					log.Printf("Error decoding token: %s", err)
+					logger.Error("Error decoding token", "error", err)
 				}
 				if len(stack) >= ml.Level {
 					buffer = append(buffer, xml.CopyToken(tok))
diff --git a/go.mod b/go.mod
index 945e272..19983a1 100644
--- a/go.mod
+++ b/go.mod
@@ -1,5 +1,11 @@
 module github.com/bmeg/sifter
 
+replace github.com/santhosh-tekuri/jsonschema/v5 v5.3.0 => github.com/bmeg/jsonschema/v5 v5.3.3
+
+go 1.21
+
+toolchain go1.21.3
+
 require (
 	github.com/akrylysov/pogreb v0.9.1
 	github.com/alecthomas/jsonschema v0.0.0-20200514014646-0366d1034a17
@@ -18,7 +24,9 @@ require (
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510
 	github.com/google/uuid v1.3.0
 	github.com/linkedin/goavro/v2 v2.10.0
+	github.com/lmittmann/tint v1.0.4
 	github.com/mattn/go-sqlite3 v1.14.16
+	github.com/rdleal/intervalst v1.3.0
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.0
 	github.com/spf13/cobra v1.6.1
 	google.golang.org/grpc v1.56.3
@@ -67,7 +75,7 @@ require (
 	github.com/klauspost/compress v1.16.0 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/kr/text v0.2.0 // indirect
-	github.com/logrusorgru/aurora v0.0.0-20200102142835-e9ef32dff381 // indirect
+	github.com/logrusorgru/aurora v0.0.0-20190428105938-cea283e61946 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.17 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
@@ -110,9 +118,3 @@ require (
 	inet.af/netaddr v0.0.0-20220811202034-502d2d690317 // indirect
 	k8s.io/apimachinery v0.26.1 // indirect
 )
-
-replace github.com/santhosh-tekuri/jsonschema/v5 v5.3.0 => github.com/bmeg/jsonschema/v5 v5.3.3
-
-go 1.21
-
-toolchain go1.21.3
diff --git a/go.sum b/go.sum
index 5434d43..1904032 100644
--- a/go.sum
+++ b/go.sum
@@ -522,10 +522,11 @@ github.com/lib/pq v1.10.2 h1:AqzbZs4ZoCBp+GtejcpCpcxM3zlSMx29dXbUSeVtJb8=
 github.com/lib/pq v1.10.2/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/linkedin/goavro/v2 v2.10.0 h1:eTBIRoInBM88gITGXYtUSqqxLTFXfOsJBiX8ZMW0o4U=
 github.com/linkedin/goavro/v2 v2.10.0/go.mod h1:UgQUb2N/pmueQYH9bfqFioWxzYCZXSfF8Jw03O5sjqA=
+github.com/lmittmann/tint v1.0.4 h1:LeYihpJ9hyGvE0w+K2okPTGUdVLfng1+nDNVR4vWISc=
+github.com/lmittmann/tint v1.0.4/go.mod h1:HIS3gSy7qNwGCj+5oRjAutErFBl4BzdQP6cJZ0NfMwE=
 github.com/logrusorgru/aurora v0.0.0-20181002194514-a7b3b318ed4e/go.mod h1:7rIyQOR62GCctdiQpZ/zOJlFyk6y+94wXzv6RNZgaR4=
+github.com/logrusorgru/aurora v0.0.0-20190428105938-cea283e61946 h1:z+WaKrgu3kCpcdnbK9YG+JThpOCd1nU5jO5ToVmSlR4=
 github.com/logrusorgru/aurora v0.0.0-20190428105938-cea283e61946/go.mod h1:7rIyQOR62GCctdiQpZ/zOJlFyk6y+94wXzv6RNZgaR4=
-github.com/logrusorgru/aurora v0.0.0-20200102142835-e9ef32dff381 h1:bqDmpDG49ZRnB5PcgP0RXtQvnMSgIF14M7CBd2shtXs=
-github.com/logrusorgru/aurora v0.0.0-20200102142835-e9ef32dff381/go.mod h1:7rIyQOR62GCctdiQpZ/zOJlFyk6y+94wXzv6RNZgaR4=
 github.com/machinebox/graphql v0.2.2/go.mod h1:F+kbVMHuwrQ5tYgU9JXlnskM8nOaFxCAEolaQybkjWA=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
@@ -661,6 +662,8 @@ github.com/prometheus/procfs v0.9.0/go.mod h1:+pB4zwohETzFnmlpe6yd2lSc+0/46IYZRB
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/quasilyte/go-consistent v0.0.0-20190521200055-c6f3937de18c/go.mod h1:5STLWrekHfjyYwxBRVRXNOSewLJ3PWfDJd1VyTS21fI=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
+github.com/rdleal/intervalst v1.3.0 h1:Ldf/5uohv4VGsrI7KQgMQ63g2EaSf9GOAhwedIoBhvc=
+github.com/rdleal/intervalst v1.3.0/go.mod h1:xO89Z6BC+LQDH+IPQQw/OESt5UADgFD41tYMUINGpxQ=
 github.com/robertkrimen/otto v0.0.0-20180617131154-15f95af6e78d/go.mod h1:xvqspoSXJTIpemEonrMDFq6XzwHYYgToXWj5eRX1OtY=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
diff --git a/graphplan/build_template.go b/graphplan/build_template.go
index 5c5e2a1..e564a3c 100644
--- a/graphplan/build_template.go
+++ b/graphplan/build_template.go
@@ -2,12 +2,12 @@ package graphplan
 
 import (
 	"fmt"
-	"log"
 	"os"
 	"path/filepath"
 	"text/template"
 
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/playbook"
 	"github.com/bmeg/sifter/task"
 )
@@ -76,7 +76,7 @@ func uniqueName(name string, used []string) string {
 	}
 }
 
-func NewGraphBuild(pb *playbook.Playbook, scriptOutDir, dataDir string) error {
+func NewGraphBuild(pb *playbook.Playbook, scriptOutDir, dataDir string, objectExclude []string) error {
 	userInputs := map[string]string{}
 	localInputs, _ := pb.PrepConfig(userInputs, filepath.Dir(pb.GetPath()))
 
@@ -96,41 +96,44 @@ func NewGraphBuild(pb *playbook.Playbook, scriptOutDir, dataDir string) error {
 		if emitName != "" {
 			for _, s := range p {
 				if s.ObjectValidate != nil {
-					schema, _ := evaluate.ExpressionString(s.ObjectValidate.Schema, task.GetConfig(), map[string]any{})
-					outdir := pb.GetDefaultOutDir()
-					outname := fmt.Sprintf("%s.%s.%s.json.gz", pb.Name, pname, emitName)
+					if !contains(s.ObjectValidate.Title, objectExclude) {
+						schema, _ := evaluate.ExpressionString(s.ObjectValidate.Schema, task.GetConfig(), map[string]any{})
+						outdir := pb.GetDefaultOutDir()
+						outname := fmt.Sprintf("%s.%s.%s.json.gz", pb.Name, pname, emitName)
 
-					outpath := filepath.Join(outdir, outname)
-					outpath, _ = filepath.Rel(scriptOutDir, outpath)
+						outpath := filepath.Join(outdir, outname)
+						outpath, _ = filepath.Rel(scriptOutDir, outpath)
 
-					schemaPath, _ := filepath.Rel(scriptOutDir, schema)
+						schemaPath, _ := filepath.Rel(scriptOutDir, schema)
 
-					_ = schemaPath
-
-					objCreate := ObjectConvertStep{Name: pname, Input: outpath, Class: s.ObjectValidate.Title, Schema: schemaPath}
-					gb.Objects = append(gb.Objects, objCreate)
+						_ = schemaPath
 
+						objCreate := ObjectConvertStep{Name: pname, Input: outpath, Class: s.ObjectValidate.Title, Schema: schemaPath}
+						gb.Objects = append(gb.Objects, objCreate)
+					}
 				}
 			}
 		}
 	}
 
 	if len(gb.Objects) > 0 {
-		log.Printf("Found %d objects", len(gb.Objects))
 		tmpl, err := template.New("graphscript").Parse(graphScript)
 		if err != nil {
 			panic(err)
 		}
 
-		outfile, err := os.Create(filepath.Join(scriptOutDir, fmt.Sprintf("%s.yaml", pb.Name)))
+		outPath := filepath.Join(scriptOutDir, fmt.Sprintf("%s.yaml", pb.Name))
+		outfile, err := os.Create(outPath)
 		if err != nil {
-			fmt.Printf("Error: %s\n", err)
+			logger.Error("File Error", "error", err)
 		}
 
+		logger.Info("Summary", "ObjectFound", len(gb.Objects), "outPath", outPath)
+
 		err = tmpl.Execute(outfile, gb)
 		outfile.Close()
 		if err != nil {
-			fmt.Printf("Error: %s\n", err)
+			logger.Error("Template Error", "error", err)
 		}
 	}
 	return nil
diff --git a/loader/dir.go b/loader/dir.go
index 94ebcac..3d62ecc 100644
--- a/loader/dir.go
+++ b/loader/dir.go
@@ -9,6 +9,8 @@ import (
 	"path/filepath"
 	"sync"
 
+	"github.com/bmeg/sifter/logger"
+
 	"google.golang.org/protobuf/encoding/protojson"
 )
 
@@ -70,7 +72,11 @@ func (s *DirDataLoader) Emit(name string, v map[string]interface{}, useName bool
 	if !ok {
 		// log.Printf("output path %s", outputPath)
 
-		j, err := os.Create(path.Join(s.dl.dir, name+".json.gz"))
+		opath := path.Join(s.dl.dir, name+".json.gz")
+
+		logger.Info("Creating emit file", "name", name, "path", opath)
+
+		j, err := os.Create(opath)
 		if err != nil {
 			return err
 		}
diff --git a/logger/logging.go b/logger/logging.go
new file mode 100644
index 0000000..bf982ca
--- /dev/null
+++ b/logger/logging.go
@@ -0,0 +1,48 @@
+package logger
+
+import (
+	"log/slog"
+	"os"
+	"time"
+
+	"github.com/lmittmann/tint"
+)
+
+var logger *slog.Logger
+
+func Init(verbose bool, json bool) {
+	level := slog.LevelInfo
+	if verbose {
+		level = slog.LevelDebug
+	}
+
+	if json {
+		logger = slog.New(
+			slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: level}),
+		)
+	} else {
+
+		logger = slog.New(
+			tint.NewHandler(os.Stderr, &tint.Options{
+				Level:      level,
+				TimeFormat: time.Kitchen,
+			}))
+
+	}
+}
+
+func Info(msg string, args ...any) {
+	logger.Info(msg, args...)
+}
+
+func Debug(msg string, args ...any) {
+	logger.Debug(msg, args...)
+}
+
+func Error(msg string, args ...any) {
+	logger.Error(msg, args...)
+}
+
+func init() {
+	Init(false, false)
+}
diff --git a/playbook/execute.go b/playbook/execute.go
index 595dfb3..2706566 100644
--- a/playbook/execute.go
+++ b/playbook/execute.go
@@ -2,10 +2,10 @@ package playbook
 
 import (
 	"fmt"
-	"log"
 	"path/filepath"
 
 	"github.com/bmeg/flame"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 	"github.com/bmeg/sifter/transform"
 )
@@ -75,9 +75,8 @@ type joinStruct struct {
 }
 
 func (pb *Playbook) Execute(task task.RuntimeTask) error {
-	log.Printf("Running playbook")
-
-	log.Printf("Inputs: %#v", task.GetConfig())
+	logger.Debug("Running playbook")
+	logger.Debug("Inputs", "config", task.GetConfig())
 
 	wf := flame.NewWorkflow()
 
@@ -90,13 +89,13 @@ func (pb *Playbook) Execute(task task.RuntimeTask) error {
 	inNodes := map[string]flame.Receiver[map[string]any]{}
 
 	for n, v := range pb.Inputs {
-		log.Printf("Setting up %s", n)
+		logger.Debug("Setting Up", "name", n)
 		s, err := v.Start(task)
 		if err == nil {
 			c := flame.AddSourceChan(wf, s)
 			outNodes[n] = c
 		} else {
-			log.Printf("Source error: %s", err)
+			logger.Error("Source error", "error", err)
 			return err
 		}
 	}
@@ -111,14 +110,14 @@ func (pb *Playbook) Execute(task task.RuntimeTask) error {
 		for i, s := range v {
 			b, err := s.Init(sub)
 			if err != nil {
-				log.Printf("Pipeline %s error: %s", k, err)
+				logger.Error("Pipeline error", "name", k, "error", err)
 				return err
 			}
 
 			procs = append(procs, b)
 
 			if mProcess, ok := b.(transform.NodeProcessor); ok {
-				log.Printf("Pipeline %s step %d: %T", k, i, b)
+				logger.Debug("PipelineSetup", "name", k, "step", i, "processor", fmt.Sprintf("%T", mProcess))
 				c := flame.AddFlatMapper(wf, mProcess.Process)
 				if lastStep != nil {
 					c.Connect(lastStep)
@@ -129,14 +128,14 @@ func (pb *Playbook) Execute(task task.RuntimeTask) error {
 						firstStep = c
 					}
 				} else {
-					log.Printf("Error setting up step")
+					logger.Error("Error setting up step")
 					//throw error?
 				}
 			} else if mProcess, ok := b.(transform.MapProcessor); ok {
-				log.Printf("Pipeline Pool %s step %d: %T", k, i, b)
+				logger.Debug("Pipeline Pool", "name", k, "step", i, "processor", b)
 				var c flame.Node[map[string]any, map[string]any]
 				if mProcess.PoolReady() {
-					log.Printf("Starting pool worker")
+					logger.Debug("Starting pool worker")
 					c = flame.AddMapperPool(wf, mProcess.Process, 4) // TODO: config pool count
 				} else {
 					c = flame.AddMapper(wf, mProcess.Process)
@@ -150,11 +149,11 @@ func (pb *Playbook) Execute(task task.RuntimeTask) error {
 						firstStep = c
 					}
 				} else {
-					log.Printf("Error setting up step")
+					logger.Error("Error setting up step")
 					//throw error?
 				}
 			} else if mProcess, ok := b.(transform.FlatMapProcessor); ok {
-				log.Printf("Pipeline flatmap %s step %d: %T", k, i, b)
+				logger.Debug("Pipeline flatmap", "name", k, "step", i, "processor", b)
 				var c flame.Node[map[string]any, map[string]any]
 				if mProcess.PoolReady() {
 					//	log.Printf("Starting pool worker")
@@ -171,11 +170,11 @@ func (pb *Playbook) Execute(task task.RuntimeTask) error {
 						firstStep = c
 					}
 				} else {
-					log.Printf("Error setting up step")
+					logger.Error("Error setting up step")
 					//throw error?
 				}
 			} else if mProcess, ok := b.(transform.StreamProcessor); ok {
-				log.Printf("Pipeline stream %s step %d: %T", k, i, b)
+				logger.Info("Pipeline stream %s step %d: %T", k, i, b)
 				c := flame.AddStreamer(wf, mProcess.Process)
 				if c != nil {
 					if lastStep != nil {
@@ -186,17 +185,17 @@ func (pb *Playbook) Execute(task task.RuntimeTask) error {
 						firstStep = c
 					}
 				} else {
-					log.Printf("Error setting up step")
+					logger.Error("Error setting up step")
 					//throw error?
 				}
 			} else if jProcess, ok := b.(transform.JoinProcessor); ok {
-				log.Printf("Pipeline Join Step")
+				logger.Debug("Pipeline Join Step")
 				c := flame.AddJoin(wf, jProcess.Process)
 				if c != nil {
 					if lastStep != nil {
 						c.ConnectLeft(lastStep)
 					} else {
-						log.Print("Join missing input")
+						logger.Debug("Join missing input")
 					}
 					joins = append(joins, joinStruct{
 						node: c,
@@ -204,11 +203,11 @@ func (pb *Playbook) Execute(task task.RuntimeTask) error {
 					})
 					lastStep = c
 					if firstStep == nil {
-						log.Printf("ERROR: Join can't be the first step")
+						logger.Error("ERROR: Join can't be the first step")
 					}
 				}
 			} else if rProcess, ok := b.(transform.ReduceProcessor); ok {
-				log.Printf("Pipeline reduce %s step %d: %T", k, i, b)
+				logger.Debug("Pipeline reduce %s step %d: %T", k, i, b)
 				wrap := reduceWrapper{rProcess}
 				k := flame.AddMapper(wf, wrap.addKeyValue)
 				r := flame.AddReduceKey(wf, rProcess.Reduce, rProcess.GetInit())
@@ -223,7 +222,7 @@ func (pb *Playbook) Execute(task task.RuntimeTask) error {
 					firstStep = k
 				}
 			} else if rProcess, ok := b.(transform.AccumulateProcessor); ok {
-				log.Printf("Pipeline accumulate %s step %d: %T", k, i, b)
+				logger.Debug("Pipeline accumulate %s step %d: %T", k, i, b)
 
 				wrap := accumulateWrapper{rProcess}
 				k := flame.AddMapper(wf, wrap.addKeyValue)
@@ -240,7 +239,7 @@ func (pb *Playbook) Execute(task task.RuntimeTask) error {
 				}
 
 			} else {
-				log.Printf("Unknown processor type")
+				logger.Info("Unknown processor type")
 			}
 		}
 		outNodes[k] = lastStep
@@ -253,25 +252,25 @@ func (pb *Playbook) Execute(task task.RuntimeTask) error {
 				src := string(*p[0].From)
 				if src == dst {
 					//TODO: more loop detection
-					log.Printf("Pipeline Loop detected in %s", dst)
+					logger.Error("Pipeline Loop detected in %s", dst)
 					return fmt.Errorf("pipeline loop detected")
 				}
 				if srcNode, ok := outNodes[src]; ok {
 					if dstNode, ok := inNodes[dst]; ok {
-						log.Printf("Connecting %s to %s ", src, dst)
+						logger.Debug("Connecting", "source", src, "dest", dst)
 						dstNode.Connect(srcNode)
 					} else {
-						log.Printf("Dest %s not found", dst)
+						logger.Error("Dest not found", "name", dst)
 					}
 				} else {
-					log.Printf("%s source %s not found", dst, src)
+					logger.Error("source not found", "dest", dst, "source", src)
 				}
 			} else {
-				log.Printf("First step of pipelines %s not 'from'", dst)
+				logger.Error("First step of pipelines not 'from'", "name", dst)
 				return fmt.Errorf("first step of pipelines %s not 'from'", dst)
 			}
 		} else {
-			log.Printf("Pipeline %s is empty", dst)
+			logger.Error("Pipeline %s is empty", dst)
 		}
 	}
 
@@ -279,17 +278,17 @@ func (pb *Playbook) Execute(task task.RuntimeTask) error {
 	for _, i := range joins {
 		r := i.proc.GetRightPipeline()
 		if srcNode, ok := outNodes[r]; ok {
-			log.Printf("Join Connect: %s", r)
+			logger.Debug("Join Connect", "source", r)
 			i.node.ConnectRight(srcNode)
 		} else {
-			log.Printf("Join source not found: %s", r)
+			logger.Error("Join source not found", "name", r)
 		}
 	}
 
 	//log.Printf("WF: %#v", wf)
 
 	wf.Start()
-	log.Printf("Workflow Started")
+	logger.Debug("Workflow Started")
 
 	wf.Wait()
 
diff --git a/transform/debug.go b/transform/debug.go
index 4da85cc..58b6e16 100644
--- a/transform/debug.go
+++ b/transform/debug.go
@@ -1,9 +1,7 @@
 package transform
 
 import (
-	"encoding/json"
-	"log"
-
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -17,13 +15,7 @@ func (ds DebugStep) Init(task task.RuntimeTask) (Processor, error) {
 }
 
 func (ds DebugStep) Process(i map[string]interface{}) []map[string]interface{} {
-	var s []byte
-	if ds.Format {
-		s, _ = json.MarshalIndent(i, "", "    ")
-	} else {
-		s, _ = json.Marshal(i)
-	}
-	log.Printf("DebugData %s: %s", ds.Label, s)
+	logger.Info("UserData", "label", ds.Label, "data", i)
 	return []map[string]any{i}
 }
 
diff --git a/transform/distinct.go b/transform/distinct.go
index ac30939..eb463cd 100644
--- a/transform/distinct.go
+++ b/transform/distinct.go
@@ -1,11 +1,11 @@
 package transform
 
 import (
-	"log"
 	"os"
 	"path/filepath"
 
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 	badger "github.com/dgraph-io/badger/v2"
 )
@@ -22,7 +22,7 @@ type distinctProcess struct {
 }
 
 func (ds DistinctStep) Init(task task.RuntimeTask) (Processor, error) {
-	log.Printf("Starting Distinct: %s", ds.Value)
+	logger.Debug("Starting Distinct: %s", ds.Value)
 	tdir := task.TempDir()
 	opts := badger.DefaultOptions(filepath.Join(tdir, "badger"))
 	opts.ValueDir = filepath.Join(tdir, "badger")
@@ -48,15 +48,15 @@ func (ds *distinctProcess) Process(i map[string]any) []map[string]any {
 			return nil
 		})
 	} else {
-		log.Printf("Distinct field error %s", err)
+		logger.Error("Distinct field error %s", err)
 	}
 	return out
 }
 
 func (ds *distinctProcess) Close() {
 	ds.db.Close()
-	log.Printf("Closing DB")
+	logger.Debug("Closing DB")
 	if err := os.RemoveAll(ds.dir); err != nil {
-		log.Printf("%s", err)
+		logger.Error("%s", err)
 	}
 }
diff --git a/transform/emit.go b/transform/emit.go
index a631129..800f93f 100644
--- a/transform/emit.go
+++ b/transform/emit.go
@@ -2,6 +2,7 @@ package transform
 
 import (
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -13,19 +14,21 @@ type EmitStep struct {
 type emitProcess struct {
 	config EmitStep
 	task   task.RuntimeTask
+	count  uint64
 }
 
 func (ts EmitStep) Init(t task.RuntimeTask) (Processor, error) {
-	return &emitProcess{ts, t}, nil
+	return &emitProcess{ts, t, 0}, nil
 }
 
-func (ts *emitProcess) Close() {}
+func (ts *emitProcess) Close() {
+	logger.Info("Emit Summary", "name", ts.config.Name, "count", ts.count)
+}
 
 func (ts *emitProcess) Process(i map[string]interface{}) []map[string]interface{} {
 	name, err := evaluate.ExpressionString(ts.config.Name, ts.task.GetConfig(), i)
 	if err == nil {
-		// log.Printf("Emitting: %s", i)
-		// log.Printf("value of UseName flag: %t", ts.config.UseName)
+		ts.count++
 		ts.task.Emit(name, i, ts.config.UseName)
 	}
 	return []map[string]any{i}
diff --git a/transform/field_process.go b/transform/field_process.go
index 51077cf..e2c40ea 100644
--- a/transform/field_process.go
+++ b/transform/field_process.go
@@ -1,9 +1,8 @@
 package transform
 
 import (
-	"log"
-
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -66,10 +65,8 @@ func (fs *fieldProcess) Process(i map[string]any) []map[string]any {
 			}
 			out = append(out, vMap)
 		} else {
-			log.Printf("Field list incorrect type: %T", v)
+			logger.Error("Field list incorrect type: %T", v)
 		}
-	} else {
-		//log.Printf("Field %s missing", fs.Field)
 	}
 	return out
 }
diff --git a/transform/filter.go b/transform/filter.go
index b383e20..c2a21c5 100644
--- a/transform/filter.go
+++ b/transform/filter.go
@@ -2,9 +2,9 @@ package transform
 
 import (
 	"fmt"
-	"log"
 
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -27,20 +27,20 @@ type filterProcessor struct {
 func (fs FilterStep) Init(task task.RuntimeTask) (Processor, error) {
 
 	if fs.Python != "" && fs.Method != "" {
-		log.Printf("Starting Filter Map: %s", fs.Python)
+		logger.Debug("Starting Filter Map: %s", fs.Python)
 		e := evaluate.GetEngine("python", task.WorkDir())
 		c, err := e.Compile(fs.Python, fs.Method)
 		if err != nil {
-			log.Printf("Compile Error: %s", err)
+			logger.Error("Compile Error: %s", err)
 		}
 		return &filterProcessor{fs, c, task}, nil
 	} else if fs.GPython != nil && fs.Method != "" {
-		log.Printf("Starting Filter Map: %s", fs.GPython)
+		logger.Debug("Starting Filter Map: %s", fs.GPython)
 		fs.GPython.SetBaseDir(task.BaseDir())
 		e := evaluate.GetEngine("gpython", task.WorkDir())
 		c, err := e.Compile(fs.GPython.String(), fs.Method)
 		if err != nil {
-			log.Printf("Compile Error: %s", err)
+			logger.Error("Compile Error: %s", err)
 		}
 		return &filterProcessor{fs, c, task}, nil
 	}
@@ -58,7 +58,7 @@ func (fs *filterProcessor) Process(row map[string]interface{}) []map[string]any
 	if fs.proc != nil {
 		out, err := fs.proc.EvaluateBool(row)
 		if err != nil {
-			log.Printf("Filter Error: %s", err)
+			logger.Error("Filter Error: %s", err)
 		}
 		if out {
 			return []map[string]any{row}
@@ -70,7 +70,7 @@ func (fs *filterProcessor) Process(row map[string]interface{}) []map[string]any
 	if fs.config.Value != "" {
 		value, err = evaluate.ExpressionString(fs.config.Value, fs.task.GetConfig(), row)
 		if err != nil {
-			log.Printf("Expression Error: %s", err)
+			logger.Error("Expression Error: %s", err)
 		}
 	} else if fs.config.Field != "" {
 		i, e := evaluate.GetJSONPath(fs.config.Field, row)
@@ -86,7 +86,7 @@ func (fs *filterProcessor) Process(row map[string]interface{}) []map[string]any
 		return []map[string]any{}
 	} else if fs.config.Check == "hasValue" {
 		if err != nil {
-			log.Printf("Field Error: %s", err)
+			logger.Error("Field Error: %s", err)
 		}
 		if err == nil && value != "" {
 			return []map[string]any{row}
@@ -94,7 +94,7 @@ func (fs *filterProcessor) Process(row map[string]interface{}) []map[string]any
 		return []map[string]any{}
 	} else if fs.config.Check == "not" {
 		if err != nil {
-			log.Printf("Field Error: %s", err)
+			logger.Error("Field Error: %s", err)
 		}
 		if err == nil && value != fs.config.Match {
 			return []map[string]any{row}
diff --git a/transform/flat_map.go b/transform/flat_map.go
index 9023383..0419759 100644
--- a/transform/flat_map.go
+++ b/transform/flat_map.go
@@ -4,9 +4,9 @@ import (
 
 	//"sync"
 	"fmt"
-	"log"
 
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -23,20 +23,20 @@ type flatMapProcess struct {
 
 func (ms *FlatMapStep) Init(task task.RuntimeTask) (Processor, error) {
 	if ms.Python != "" {
-		log.Printf("Init Map: %s", ms.Python)
+		logger.Debug("Init Map: %s", ms.Python)
 		e := evaluate.GetEngine("python", task.WorkDir())
 		c, err := e.Compile(ms.Python, ms.Method)
 		if err != nil {
-			log.Printf("Compile Error: %s", err)
+			logger.Error("Compile Error: %s", err)
 		}
 		return &flatMapProcess{ms, c}, nil
 	} else if ms.GPython != nil {
-		log.Printf("Init Map: %s", ms.GPython)
+		logger.Debug("Init Map: %s", ms.GPython)
 		ms.GPython.SetBaseDir(task.BaseDir())
 		e := evaluate.GetEngine("gpython", task.WorkDir())
 		c, err := e.Compile(ms.GPython.String(), ms.Method)
 		if err != nil {
-			log.Printf("Compile Error: %s", err)
+			logger.Error("Compile Error: %s", err)
 		}
 		return &flatMapProcess{ms, c}, nil
 	}
@@ -50,15 +50,14 @@ func (mp *flatMapProcess) PoolReady() bool {
 func (mp *flatMapProcess) Process(i map[string]interface{}) []map[string]interface{} {
 	o, err := mp.proc.EvaluateArray(i)
 	if err != nil {
-		log.Printf("FlatMap Step error: %s", err)
+		logger.Error("FlatMap Step error: %s", err)
 	}
-	//log.Printf("Flatmap out: %#v", o)
 	out := []map[string]any{}
 	for _, i := range o {
 		if m, ok := i.(map[string]any); ok {
 			out = append(out, m)
 		} else {
-			log.Printf("Flatmap output error: %#v", i)
+			logger.Error("Flatmap output error: %#v", i)
 		}
 	}
 	return out
diff --git a/transform/graph_build.go b/transform/graph_build.go
index 7ed31b1..6f657c1 100644
--- a/transform/graph_build.go
+++ b/transform/graph_build.go
@@ -1,11 +1,10 @@
 package transform
 
 import (
-	"log"
-
 	schema "github.com/bmeg/jsonschemagraph/util"
 	"github.com/bmeg/sifter/config"
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -51,11 +50,11 @@ func (ts GraphBuildStep) Init(task task.RuntimeTask) (Processor, error) {
 	if ts.EdgeFix != nil {
 		if ts.EdgeFix.GPython != nil {
 			ts.EdgeFix.GPython.SetBaseDir(task.BaseDir())
-			log.Printf("Init Map: %s", ts.EdgeFix.GPython)
+			logger.Debug("Init Map: %s", ts.EdgeFix.GPython)
 			e := evaluate.GetEngine("gpython", task.WorkDir())
 			c, err := e.Compile(ts.EdgeFix.GPython.String(), ts.EdgeFix.Method)
 			if err != nil {
-				log.Printf("Compile Error: %s", err)
+				logger.Error("Compile Error: %s", err)
 			}
 			edgeFix = c
 		}
@@ -88,7 +87,7 @@ func (ts *graphBuildProcess) Process(i map[string]interface{}) []map[string]inte
 			if j.Vertex != nil {
 				err := ts.task.Emit("vertex", ts.vertexToMap(j.Vertex), false)
 				if err != nil {
-					log.Printf("Emit Error: %s", err)
+					logger.Error("Emit Error: %s", err)
 				}
 			} else if j.OutEdge != nil || j.InEdge != nil {
 				var edge *schema.Edge
@@ -108,13 +107,13 @@ func (ts *graphBuildProcess) Process(i map[string]interface{}) []map[string]inte
 					}
 					err := ts.task.Emit("edge", edgeData, false)
 					if err != nil {
-						log.Printf("Emit Error: %s", err)
+						logger.Error("Emit Error: %s", err)
 					}
 				}
 			}
 		}
 	} else {
-		log.Printf("Graphbuild %s error : %s", ts.config.Title, err)
+		logger.Error("Graphbuild %s error : %s", ts.config.Title, err)
 	}
 
 	return out
diff --git a/transform/hash.go b/transform/hash.go
index 7ff9d7a..a2cb45e 100644
--- a/transform/hash.go
+++ b/transform/hash.go
@@ -32,7 +32,7 @@ func (fs HashStep) Init(task task.RuntimeTask) (Processor, error) {
 	} else if fs.Method == "md5" || fs.Method == "" {
 		return &hashProcessor{fs, md5.New(), task}, nil
 	}
-	return nil, fmt.Errorf("Hash method %s not found", fs.Method)
+	return nil, fmt.Errorf("hash method %s not found", fs.Method)
 }
 
 func (fs *hashProcessor) Process(i map[string]any) []map[string]any {
diff --git a/transform/interface.go b/transform/interface.go
index ffe1aa9..633da96 100644
--- a/transform/interface.go
+++ b/transform/interface.go
@@ -55,31 +55,32 @@ type Transform interface {
 }
 
 type Step struct {
-	Docs           string              `json:"docs"`
-	From           *FromStep           `json:"from"`
-	Split          *SplitStep          `json:"split"`
-	FieldParse     *FieldParseStep     `json:"fieldParse" jsonschema_description:"fieldParse to run"`
-	FieldType      *FieldTypeStep      `json:"fieldType" jsonschema_description:"Change type of a field (ie string -> integer)"`
-	ObjectValidate *ObjectValidateStep `json:"objectValidate" jsonschema_description:"Validate a JSON schema based object"`
-	Emit           *EmitStep           `json:"emit" jsonschema_description:"Write to unstructured JSON file"`
-	Filter         *FilterStep         `json:"filter"`
-	Clean          *CleanStep          `json:"clean"`
-	Debug          *DebugStep          `json:"debug" jsonschema_description:"Print message contents to stdout"`
-	RegexReplace   *RegexReplaceStep   `json:"regexReplace"`
-	Project        *ProjectStep        `json:"project" jsonschema_description:"Run a projection mapping message"`
-	Map            *MapStep            `json:"map" jsonschema_description:"Apply a single function to all records"`
-	Plugin         *PluginStep         `json:"plugin"`
-	FlatMap        *FlatMapStep        `json:"flatmap" jsonschema_description:"Apply a single function to all records, flatten list responses"`
-	Reduce         *ReduceStep         `json:"reduce"`
-	Distinct       *DistinctStep       `json:"distinct"`
-	FieldProcess   *FieldProcessStep   `json:"fieldProcess" jsonschema_description:"Take an array field from a message and run in child transform"`
-	DropNull       *DropNullStep       `json:"dropNull"`
-	Lookup         *LookupStep         `json:"lookup"`
-	Hash           *HashStep           `json:"hash"`
-	GraphBuild     *GraphBuildStep     `json:"graphBuild"`
-	TableWrite     *TableWriteStep     `json:"tableWrite"`
-	Accumulate     *AccumulateStep     `json:"accumulate"`
-	UUID           *UUIDStep           `json:"uuid"`
+	Docs              string              `json:"docs"`
+	From              *FromStep           `json:"from"`
+	Split             *SplitStep          `json:"split"`
+	FieldParse        *FieldParseStep     `json:"fieldParse" jsonschema_description:"fieldParse to run"`
+	FieldType         *FieldTypeStep      `json:"fieldType" jsonschema_description:"Change type of a field (ie string -> integer)"`
+	ObjectValidate    *ObjectValidateStep `json:"objectValidate" jsonschema_description:"Validate a JSON schema based object"`
+	Emit              *EmitStep           `json:"emit" jsonschema_description:"Write to unstructured JSON file"`
+	Filter            *FilterStep         `json:"filter"`
+	Clean             *CleanStep          `json:"clean"`
+	Debug             *DebugStep          `json:"debug" jsonschema_description:"Print message contents to stdout"`
+	RegexReplace      *RegexReplaceStep   `json:"regexReplace"`
+	Project           *ProjectStep        `json:"project" jsonschema_description:"Run a projection mapping message"`
+	Map               *MapStep            `json:"map" jsonschema_description:"Apply a single function to all records"`
+	Plugin            *PluginStep         `json:"plugin"`
+	FlatMap           *FlatMapStep        `json:"flatmap" jsonschema_description:"Apply a single function to all records, flatten list responses"`
+	Reduce            *ReduceStep         `json:"reduce"`
+	Distinct          *DistinctStep       `json:"distinct"`
+	FieldProcess      *FieldProcessStep   `json:"fieldProcess" jsonschema_description:"Take an array field from a message and run in child transform"`
+	DropNull          *DropNullStep       `json:"dropNull"`
+	Lookup            *LookupStep         `json:"lookup"`
+	IntervalIntersect *IntervalStep       `json:"intervalIntersect"`
+	Hash              *HashStep           `json:"hash"`
+	GraphBuild        *GraphBuildStep     `json:"graphBuild"`
+	TableWrite        *TableWriteStep     `json:"tableWrite"`
+	Accumulate        *AccumulateStep     `json:"accumulate"`
+	UUID              *UUIDStep           `json:"uuid"`
 }
 
 type Pipe []Step
diff --git a/transform/lookup.go b/transform/lookup.go
index f82fddc..0d1a506 100644
--- a/transform/lookup.go
+++ b/transform/lookup.go
@@ -4,7 +4,6 @@ import (
 	"compress/gzip"
 	"fmt"
 	"io"
-	"log"
 	"os"
 	"strings"
 
@@ -13,6 +12,7 @@ import (
 
 	"github.com/bmeg/golib"
 	"github.com/bmeg/sifter/config"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 
 	"github.com/bmeg/sifter/evaluate"
@@ -116,7 +116,7 @@ func (tr *LookupStep) GetConfigFields() []config.Variable {
 }
 
 func (tp *lookupProcess) Close() {
-	log.Printf("Table hits: %d misses: %d", tp.hitCount, tp.missCount)
+	logger.Info("Table summary", "hit", tp.hitCount, "miss", tp.missCount)
 }
 
 func (tp *lookupProcess) Process(row map[string]interface{}) []map[string]interface{} {
@@ -224,7 +224,7 @@ func (tsv *TSVTable) open(task task.RuntimeTask) (lookupTable, error) {
 	} else if s.IsDir() {
 		return nil, fmt.Errorf("file not found: %s", inputPath)
 	}
-	log.Printf("Loading Translation file: %s", inputPath)
+	logger.Debug("Loading Translation file", "path", inputPath)
 
 	var inputStream io.Reader
 	if gfile, err := os.Open(inputPath); err == nil {
@@ -238,7 +238,7 @@ func (tsv *TSVTable) open(task task.RuntimeTask) (lookupTable, error) {
 			inputStream = gfile
 		}
 	} else if err != nil {
-		log.Printf("Error loading table: %s", err)
+		logger.Error("Error loading table", "error", err)
 		return nil, err
 	}
 
@@ -262,7 +262,7 @@ func (tsv *TSVTable) open(task task.RuntimeTask) (lookupTable, error) {
 
 	lines, err := tsvReader.ReadAll()
 	if err != nil {
-		log.Printf("Error loading lookup table: %s", err)
+		logger.Error("Error loading lookup table", "error", err)
 		return nil, err
 	}
 
@@ -276,7 +276,7 @@ func (tsv *TSVTable) open(task task.RuntimeTask) (lookupTable, error) {
 			tp.table[row[tp.colmap[tsv.Key]]] = row
 		}
 	}
-	log.Printf("tableLookup loaded %d values from %s", len(tp.table), inputPath)
+	logger.Debug("tableLookup summary", "count", len(tp.table), "path", inputPath)
 	return tp, nil
 }
 
@@ -310,7 +310,7 @@ func (jf *JSONTable) open(task task.RuntimeTask) (lookupTable, error) {
 	} else if s.IsDir() {
 		return nil, fmt.Errorf("file not found: %s", inputPath)
 	}
-	log.Printf("Loading Translation file: %s", inputPath)
+	logger.Debug("Loading Translation file", "path", inputPath)
 
 	var inputStream chan []byte
 	if strings.HasSuffix(inputPath, ".gz") {
@@ -337,7 +337,7 @@ func (jf *JSONTable) open(task task.RuntimeTask) (lookupTable, error) {
 			}
 		}
 	}
-	log.Printf("jsonLookup loaded %d values from %s", len(jp.table), inputPath)
+	logger.Debug("jsonLookup summary", "count", len(jp.table), "path", inputPath)
 
 	return jp, nil
 }
diff --git a/transform/lookup_interval.go b/transform/lookup_interval.go
new file mode 100644
index 0000000..8ebd661
--- /dev/null
+++ b/transform/lookup_interval.go
@@ -0,0 +1,205 @@
+package transform
+
+import (
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/bmeg/golib"
+	"github.com/bmeg/sifter/config"
+	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
+	"github.com/bmeg/sifter/task"
+
+	"github.com/rdleal/intervalst/interval"
+)
+
+type NamedPosition struct {
+	Name string
+	Pos  int64
+}
+
+func NamedCmp(x, y NamedPosition) int {
+	v := strings.Compare(x.Name, y.Name)
+	if v != 0 {
+		return v
+	}
+	return int(x.Pos - y.Pos)
+}
+
+func NumCmp(x, y int64) int {
+	return int(x - y)
+}
+
+type IntervalStep struct {
+	Match string        `json:"match"`
+	Start string        `json:"start"`
+	End   string        `json:"end"`
+	Field string        `json:"field"`
+	JSON  *JSONInterval `json:"json"`
+	//TSV   *TSVInterval      `json:"tsv"`
+}
+
+type JSONInterval struct {
+	Match string `json:"match"`
+	Input string `json:"input"`
+	Start string `json:"start"`
+	End   string `json:"end"`
+}
+
+type TSVInterval struct {
+	Start string `json:"start"`
+	End   string `json:"end"`
+}
+
+type intervalProcess struct {
+	config    *IntervalStep
+	table     map[string]*interval.SearchTree[map[string]any, int64]
+	hitCount  int
+	missCount int
+	//table  map[string][]string
+}
+
+func toInt64(v any) int64 {
+	switch t := v.(type) {
+	case int:
+		return int64(t)
+	case int8:
+		return int64(t)
+	case int16:
+		return int64(t)
+	case int32:
+		return int64(t)
+	case int64:
+		return int64(t)
+	case uint:
+		return int64(t)
+	case uint8:
+		return int64(t)
+	case uint16:
+		return int64(t)
+	case uint32:
+		return int64(t)
+	case uint64:
+		return int64(t)
+	case float32:
+		return int64(t)
+	case float64:
+		return int64(t)
+	case string:
+		i, err := strconv.ParseInt(t, 10, 64)
+		if err == nil {
+			return i
+		}
+	}
+	//fmt.Printf("Issues: %T\n", v)
+	return 0
+}
+
+func (tr *IntervalStep) GetConfigFields() []config.Variable {
+	out := []config.Variable{}
+	if tr.JSON != nil && tr.JSON.Input != "" {
+		for _, s := range evaluate.ExpressionIDs(tr.JSON.Input) {
+			out = append(out, config.Variable{Type: config.File, Name: config.TrimPrefix(s)})
+		}
+	}
+	return out
+}
+
+func (tr *IntervalStep) Init(task task.RuntimeTask) (Processor, error) {
+	if tr.JSON != nil {
+
+		tableMap := map[string]*interval.SearchTree[map[string]any, int64]{}
+		var err error
+
+		inputPath, err := evaluate.ExpressionString(tr.JSON.Input, task.GetConfig(), nil)
+		logger.Debug("Opening interval file", "path", inputPath, "input", tr.JSON.Input, "config", task.GetConfig())
+		if err != nil {
+			return nil, err
+		}
+
+		var inputStream chan []byte
+		if strings.HasSuffix(inputPath, ".gz") {
+			inputStream, err = golib.ReadGzipLines(inputPath)
+		} else {
+			inputStream, err = golib.ReadFileLines(inputPath)
+		}
+		if err != nil {
+			return nil, err
+		}
+		count := 0
+		for line := range inputStream {
+			if len(line) > 0 {
+				row := map[string]any{}
+				err := json.Unmarshal(line, &row)
+				if err != nil {
+					return nil, err
+				}
+				if match, ok := row[tr.JSON.Match]; ok {
+					if matchString, ok := match.(string); ok {
+						var table *interval.SearchTree[map[string]any, int64]
+						if table, ok = tableMap[matchString]; !ok {
+							table = interval.NewSearchTree[map[string]any](NumCmp)
+							tableMap[matchString] = table
+						}
+
+						if startString, ok := row[tr.JSON.Start]; ok {
+							start := toInt64(startString)
+							if endString, ok := row[tr.JSON.End]; ok {
+								end := toInt64(endString)
+								if count < 10 {
+									logger.Debug("intervalIntersect load", "match", matchString, "start", start, "end", end)
+								}
+								table.Insert(start, end, row)
+								count++
+							}
+						}
+					}
+				}
+			}
+		}
+
+		logger.Debug("intervalIntersect loaded", "values", count)
+
+		return &intervalProcess{config: tr, table: tableMap}, err
+	}
+	return nil, fmt.Errorf("interval input not defined")
+}
+
+func (tp *intervalProcess) Close() {
+	logger.Info("Table summary", "hit", tp.hitCount, "miss", tp.missCount)
+}
+
+func (tp *intervalProcess) Process(row map[string]interface{}) []map[string]interface{} {
+
+	if match, ok := row[tp.config.Match]; ok {
+		if matchString, ok := match.(string); ok {
+			if table, ok := tp.table[matchString]; ok {
+				if start, ok := row[tp.config.Start]; ok {
+					startInt := toInt64(start)
+					if end, ok := row[tp.config.End]; ok {
+						endInt := toInt64(end)
+						vals, found := table.AllIntersections(
+							startInt,
+							endInt,
+						)
+						if found {
+							row[tp.config.Field] = vals
+							tp.hitCount++
+						} else {
+							if tp.missCount < 10 {
+								logger.Debug("intervalIntersect miss",
+									"match", matchString, "start", startInt, "end", endInt)
+							}
+							row[tp.config.Field] = []any{}
+							tp.missCount++
+						}
+					}
+				}
+			}
+		}
+	}
+
+	return []map[string]any{row}
+}
diff --git a/transform/lookup_pipeline.go b/transform/lookup_pipeline.go
index 97dbbc0..5b1ccb1 100644
--- a/transform/lookup_pipeline.go
+++ b/transform/lookup_pipeline.go
@@ -1,8 +1,7 @@
 package transform
 
 import (
-	"log"
-
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -25,10 +24,10 @@ func (pls *pipelineLookupProcess) Process(left chan map[string]any, right chan m
 				vals[kStr] = i
 			}
 		} else {
-			log.Printf("Missing key %s : %#v", pls.config.Pipeline.Key, i)
+			logger.Info("Missing key", "key", pls.config.Pipeline.Key)
 		}
 	}
-	log.Printf("Pipeline lookup loaded %d values", len(vals))
+	logger.Info("Pipeline lookup", "size", len(vals))
 	table := recordTable{vals, pls.config.Pipeline.Value}
 	lk := lookupProcess{pls.config, &table, pls.task.GetConfig(), 0, 0}
 
diff --git a/transform/mapping.go b/transform/mapping.go
index 1496010..36f30c3 100644
--- a/transform/mapping.go
+++ b/transform/mapping.go
@@ -4,9 +4,9 @@ import (
 
 	//"sync"
 	"fmt"
-	"log"
 
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -23,20 +23,20 @@ type mapProcess struct {
 
 func (ms *MapStep) Init(task task.RuntimeTask) (Processor, error) {
 	if ms.Python != "" {
-		log.Printf("Init Map: %s", ms.Python)
+		logger.Debug("Init Map: %s", ms.Python)
 		e := evaluate.GetEngine("python", task.WorkDir())
 		c, err := e.Compile(ms.Python, ms.Method)
 		if err != nil {
-			log.Printf("Compile Error: %s", err)
+			logger.Error("Compile Error: %s", err)
 		}
 		return &mapProcess{ms, c}, nil
 	} else if ms.GPython != nil {
-		log.Printf("Init Map: %s", ms.GPython)
+		logger.Debug("Init Map: %s", ms.GPython)
 		ms.GPython.SetBaseDir(task.BaseDir())
 		e := evaluate.GetEngine("gpython", task.WorkDir())
 		c, err := e.Compile(ms.GPython.String(), ms.Method)
 		if err != nil {
-			log.Printf("Compile Error: %s", err)
+			logger.Error("Compile Error: %s", err)
 		}
 		return &mapProcess{ms, c}, nil
 	}
@@ -50,7 +50,7 @@ func (mp *mapProcess) PoolReady() bool {
 func (mp *mapProcess) Process(i map[string]interface{}) map[string]interface{} {
 	out, err := mp.proc.Evaluate(i)
 	if err != nil {
-		log.Printf("Map Step error: %s", err)
+		logger.Error("Map Step error: %s", err)
 	}
 	return out
 }
diff --git a/transform/object_validate.go b/transform/object_validate.go
index 08838cc..13f8cdb 100644
--- a/transform/object_validate.go
+++ b/transform/object_validate.go
@@ -1,13 +1,12 @@
 package transform
 
 import (
-	"encoding/json"
 	"fmt"
-	"log"
 
 	schema "github.com/bmeg/jsonschemagraph/util"
 	"github.com/bmeg/sifter/config"
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 	"github.com/santhosh-tekuri/jsonschema/v5"
 	_ "github.com/santhosh-tekuri/jsonschema/v5/httploader" // setup the httploader for the jsonschema checker
@@ -20,12 +19,13 @@ type ObjectValidateStep struct {
 }
 
 type objectProcess struct {
-	config     ObjectValidateStep
-	task       task.RuntimeTask
-	className  string
-	schema     schema.GraphSchema
-	class      *jsonschema.Schema
-	errorCount int
+	config      ObjectValidateStep
+	task        task.RuntimeTask
+	className   string
+	schema      schema.GraphSchema
+	class       *jsonschema.Schema
+	errorCount  int
+	objectCount uint64
 }
 
 func (ts ObjectValidateStep) Init(task task.RuntimeTask) (Processor, error) {
@@ -40,14 +40,14 @@ func (ts ObjectValidateStep) Init(task task.RuntimeTask) (Processor, error) {
 	}
 	if ts.Title != "" {
 		if cls := sc.GetClass(ts.Title); cls != nil {
-			return &objectProcess{ts, task, ts.Title, sc, cls, 0}, nil
+			return &objectProcess{ts, task, ts.Title, sc, cls, 0, 0}, nil
 		}
 		return nil, fmt.Errorf("class %s not found", ts.Title)
 	}
 	if ts.URI != "" {
 		uri := path + "/" + ts.URI
 		if cls := sc.GetClass(uri); cls != nil {
-			return &objectProcess{ts, task, cls.Title, sc, cls, 0}, nil
+			return &objectProcess{ts, task, cls.Title, sc, cls, 0, 0}, nil
 		}
 		return nil, fmt.Errorf("uri %s not found", uri)
 	}
@@ -69,18 +69,19 @@ func (ts *objectProcess) PoolReady() bool {
 }
 
 func (ts *objectProcess) Process(i map[string]interface{}) []map[string]interface{} {
+	ts.objectCount++
 	out, err := ts.schema.CleanAndValidate(ts.class, i)
 	if err == nil {
 		return []map[string]any{out}
 	}
 	//if ts.errorCount < 10 {
-	data, _ := json.Marshal(i)
-	log.Printf("validate %s error: %s on %s", ts.className, err, data)
+	logger.Error("validation error", "className", ts.className, "error", err)
+	logger.Debug("validation data", "data", i)
 	//}
 	ts.errorCount++
 	return []map[string]any{}
 }
 
 func (ts *objectProcess) Close() {
-	log.Printf("Total incorrect rows: %d", ts.errorCount)
+	logger.Info("Validation Summary", "name", ts.task.GetName(), "class", ts.className, "errorCount", ts.errorCount, "validationCount", ts.objectCount)
 }
diff --git a/transform/plugin.go b/transform/plugin.go
index cfe3b18..b16b817 100644
--- a/transform/plugin.go
+++ b/transform/plugin.go
@@ -4,11 +4,11 @@ import (
 	"bufio"
 	"encoding/json"
 	"io"
-	"log"
 	"os"
 	"os/exec"
 	"sync"
 
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 	"github.com/google/shlex"
 )
@@ -30,7 +30,7 @@ func (ps *pluginProcess) Process(in chan map[string]any, out chan map[string]any
 
 	cmdLine, err := shlex.Split(ps.config.CommandLine)
 	if err != nil {
-		log.Printf("Plugin Error: %s", err)
+		logger.Error("Plugin Error: %s", err)
 	} else {
 		workdir := ps.task.BaseDir()
 		cmd := exec.Command(cmdLine[0], cmdLine[1:]...)
@@ -38,10 +38,10 @@ func (ps *pluginProcess) Process(in chan map[string]any, out chan map[string]any
 		stdin, _ := cmd.StdinPipe()
 		stdout, _ := cmd.StdoutPipe()
 		cmd.Stderr = os.Stderr
-		log.Printf("Starting: %#v", cmd)
+		logger.Debug("Starting: %#v", cmd)
 		err := cmd.Start()
 		if err != nil {
-			log.Printf("plugin exec error: %s", err)
+			logger.Error("plugin exec error: %s", err)
 		}
 
 		go func() {
@@ -63,7 +63,7 @@ func (ps *pluginProcess) Process(in chan map[string]any, out chan map[string]any
 			for err == nil {
 				line, isPrefix, err = reader.ReadLine()
 				if err != nil && err != io.EOF {
-					log.Printf("plugin (%s) input error: %s", ps.config.CommandLine, err)
+					logger.Error("plugin (%s) input error: %s", ps.config.CommandLine, err)
 				}
 				ln = append(ln, line...)
 				if !isPrefix {
@@ -73,8 +73,8 @@ func (ps *pluginProcess) Process(in chan map[string]any, out chan map[string]any
 						if err == nil {
 							out <- row
 						} else {
-							log.Printf("plugin output error: %s", err)
-							log.Printf("unmarshalled line: %s", ln)
+							logger.Error("plugin output error: %s", err)
+							logger.Error("unmarshalled line: %s", ln)
 						}
 						ln = []byte{}
 					}
@@ -82,7 +82,7 @@ func (ps *pluginProcess) Process(in chan map[string]any, out chan map[string]any
 			}
 			wg.Done()
 		}()
-		log.Printf("plugin has exited: %s\n", ps.config.CommandLine)
+		logger.Debug("plugin has exited: %s\n", ps.config.CommandLine)
 		wg.Wait()
 	}
 }
diff --git a/transform/reduce.go b/transform/reduce.go
index 8cabcaa..005804f 100644
--- a/transform/reduce.go
+++ b/transform/reduce.go
@@ -2,9 +2,9 @@ package transform
 
 import (
 	"fmt"
-	"log"
 
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -23,22 +23,22 @@ type reduceProcess struct {
 
 func (ms *ReduceStep) Init(t task.RuntimeTask) (Processor, error) {
 	if ms.Python != "" {
-		log.Printf("ReduceInit: %s", ms.InitData)
-		log.Printf("Reduce: %s", ms.Python)
+		logger.Debug("ReduceInit: %s", ms.InitData)
+		logger.Debug("Reduce: %s", ms.Python)
 		e := evaluate.GetEngine("python", t.WorkDir())
 		c, err := e.Compile(ms.Python, ms.Method)
 		if err != nil {
-			log.Printf("Compile Error: %s", err)
+			logger.Error("Compile Error: %s", err)
 		}
 		return &reduceProcess{ms, c}, nil
 	} else if ms.GPython != nil {
 		ms.GPython.SetBaseDir(t.BaseDir())
-		log.Printf("ReduceInit: %s", ms.InitData)
-		log.Printf("Reduce: %s", ms.GPython)
+		logger.Debug("ReduceInit: %s", ms.InitData)
+		logger.Debug("Reduce: %s", ms.GPython)
 		e := evaluate.GetEngine("gpython", t.WorkDir())
 		c, err := e.Compile(ms.GPython.String(), ms.Method)
 		if err != nil {
-			log.Printf("Compile Error: %s", err)
+			logger.Error("Compile Error: %s", err)
 		}
 		return &reduceProcess{ms, c}, nil
 	}
@@ -62,7 +62,7 @@ func (rp *reduceProcess) GetKey(i map[string]any) string {
 			return xStr
 		}
 	} else {
-		log.Printf("Missing field in reduce")
+		logger.Info("Missing field in reduce: %s", rp.config.Field)
 	}
 	return ""
 }
@@ -70,7 +70,7 @@ func (rp *reduceProcess) GetKey(i map[string]any) string {
 func (rp *reduceProcess) Reduce(key string, a map[string]any, b map[string]any) map[string]any {
 	out, err := rp.proc.Evaluate(a, b)
 	if err != nil {
-		log.Printf("Reduce Error: %s", err)
+		logger.Error("Reduce Error: %s", err)
 	}
 	return out
 }
diff --git a/transform/table_write.go b/transform/table_write.go
index 3e491bc..2c36d62 100644
--- a/transform/table_write.go
+++ b/transform/table_write.go
@@ -5,12 +5,12 @@ import (
 	"encoding/csv"
 	"encoding/json"
 	"io"
-	"log"
 	"os"
 	"path/filepath"
 	"strings"
 
 	"github.com/bmeg/sifter/evaluate"
+	"github.com/bmeg/sifter/logger"
 	"github.com/bmeg/sifter/task"
 )
 
@@ -41,7 +41,7 @@ func (tw *TableWriteStep) Init(task task.RuntimeTask) (Processor, error) {
 		return nil, err
 	}
 	outputPath := filepath.Join(task.OutDir(), output)
-	log.Printf("Starting TableWriter to %s", outputPath)
+	logger.Info("Starting TableWriter to %s", outputPath)
 
 	te := tableWriteProcess{}
 	te.handle, _ = os.Create(outputPath)
@@ -70,7 +70,7 @@ func (tw *TableWriteStep) GetOutputs(task task.RuntimeTask) []string {
 		return []string{}
 	}
 	outputPath := filepath.Join(task.OutDir(), output)
-	log.Printf("table output %s %s", task.OutDir(), output)
+	logger.Debug("table output %s %s", task.OutDir(), output)
 	return []string{outputPath}
 }
 
@@ -95,7 +95,7 @@ func (tp *tableWriteProcess) Process(i map[string]any) map[string]any {
 }
 
 func (tp *tableWriteProcess) Close() {
-	log.Printf("Closing tableWriter: %s", tp.config.Output)
+	logger.Debug("Closing tableWriter: %s", tp.config.Output)
 	tp.writer.Flush()
 	tp.out.Close()
 	tp.handle.Close()