api

type CaptureLoggerAdapter ¶

type CaptureLoggerAdapter struct {
	// contains filtered or unexported fields
}

CaptureLoggerAdapter is a logger which captures all logs. This logger is mostly useful for testing logging.

func NewCaptureLogger() *CaptureLoggerAdapter

func (c *CaptureLoggerAdapter) Captured() map[LogLevel][]CapturedMessage

func (c *CaptureLoggerAdapter) Debug(msg string, fields LogFields)

func (c *CaptureLoggerAdapter) Error(msg string, err error, fields LogFields)

func (c *CaptureLoggerAdapter) Has(msg CapturedMessage) bool

func (c *CaptureLoggerAdapter) HasError(err error) bool

func (c *CaptureLoggerAdapter) Info(msg string, fields LogFields)

func (c *CaptureLoggerAdapter) PrintCaptured(t Logfer)

func (c *CaptureLoggerAdapter) Trace(msg string, fields LogFields)

func (c *CaptureLoggerAdapter) With(fields LogFields) LoggerAdapter

type CapturedMessage ¶

type CapturedMessage struct {
	Level  LogLevel
	Time   time.Time
	Fields LogFields
	Msg    string
	Err    error
}

func (c CapturedMessage) ContentEquals(other CapturedMessage) bool

type Combiner ¶

type Combiner func(key string, values Iterator, emit Emitter) error

Combiner is a function that aggregates values for a specific key.

Arguments:

• key: The intermediate key (e.g., a word in WordCount).
• values: An Iterator allowing sequential access to all values for this key.

Returns:

• error: Any error encountered during aggregation.

type Emitter ¶

type Emitter func(key, value string) error

Emitter is the callback function passed to the Mapper.

It allows the Mapper to "emit" intermediate key-value pairs.

type Filesystem ¶

type Filesystem interface {
	// Glob returns a list of file paths matching the specified pattern.
	//
	// The pattern syntax is implementation-specific but generally follows shell
	// globbing rules (e.g., "input/*.txt" or "s3://bucket/data/2023-*").
	Glob(pattern string) ([]string, error)

	// Size returns the size of the file in bytes.
	//
	// This is primarily used by the Master to calculate Input Splits (assigning
	// byte ranges to workers) without downloading the file content.
	Size(filename string) (int64, error)

	// Open opens the named file for reading.
	//
	// It returns an io.ReadSeekCloser, allowing the caller to efficiently Seek()
	// to a specific offset. This is crucial for workers to read only their
	// assigned partition (chunk) of a large input file.
	Open(filename string) (io.ReadSeekCloser, error)

	// Create opens the named file for writing.
	//
	// If the file already exists, it should be truncated. This is used by
	// workers to write intermediate shuffle data and final MapReduce outputs.
	Create(filename string) (io.WriteCloser, error)

	// Delete removes the named file from the storage system.
	//
	// This is used for cleaning up intermediate files after the Reduce phase
	// completes or for removing temporary artifacts upon job failure.
	Delete(filename string) error

	// Abs returns an absolute representation of the path.
	Abs(path string) (string, error)
}

Filesystem abstracts the underlying storage layer, providing a unified API for interacting with local disks, distributed file systems (like HDFS), or object storage (like S3).

Implementations of this interface must be safe for concurrent use by multiple goroutines.

type Iterator ¶

type Iterator interface {
	// Next returns the next value in the stream.
	//
	// It returns false as the second return value when stream is exhausted.
	Next() (string, bool)

	// Close cleans up any resources (file handles, network connections)
	// associated with the iterator.
	io.Closer
}

Iterator represents a stream of values associated with a specific key.

It allows the Reducer to process data larger than available RAM by strictly processing one item at a time.

type LogFields ¶

type LogFields map[string]any

LogFields is the logger's key-value list of fields.

func (l LogFields) Add(newFields LogFields) LogFields

func (l LogFields) Copy() LogFields

type LogLevel ¶

type LogLevel uint

const (
	TraceLogLevel LogLevel = iota + 1
	DebugLogLevel
	InfoLogLevel
	ErrorLogLevel
)

type Logfer ¶

type Logfer interface {
	Logf(format string, a ...any)
}

type LoggerAdapter ¶

type LoggerAdapter interface {
	Error(msg string, err error, fields LogFields)
	Info(msg string, fields LogFields)
	Debug(msg string, fields LogFields)
	Trace(msg string, fields LogFields)
	With(fields LogFields) LoggerAdapter
}

LoggerAdapter is an interface, that you need to implement to support Shard logging.

You can use StdLoggerAdapter as a reference implementation.

func NewStdLogger(debug, trace bool) LoggerAdapter

func NewStdLoggerWithOut(out io.Writer, debug bool, trace bool) LoggerAdapter

type Mapper ¶

type Mapper func(key, value string, emit Emitter) error

Mapper is a function that transforms raw input into intermediate key-value pairs.

Arguments:

• key: The identifier for the input data (e.g., filename, offset).
• value: The raw data content (e.g., line of text, entire file).
• emit: A callback to output 0 or more key-value pairs.

Returns:

• error: Any error encountered during aggregation.

type NopLogger ¶

type NopLogger struct{}

NopLogger is a logger which discards all logs.

func (NopLogger) Debug(msg string, fields LogFields)

func (NopLogger) Error(msg string, err error, fields LogFields)

func (NopLogger) Info(msg string, fields LogFields)

func (NopLogger) Trace(msg string, fields LogFields)

func (l NopLogger) With(fields LogFields) LoggerAdapter

type Partitioner ¶

type Partitioner func(key string, numPartitions int) int

Partitioner determines which Reduce task should process a specific key.

Arguments:

• key: The intermediate key to hash.
• numPartitions: The total number of available Reduce tasks.

Returns:

• int: The partition index (0 to numPartitions-1).

type Reducer ¶

type Reducer func(key string, values Iterator, emit Emitter) error

Reducer is a function that aggregates values for a specific key.

Arguments:

• key: The intermediate key (e.g., a word in WordCount).
• values: An Iterator allowing sequential access to all values for this key.

Returns:

• error: Any error encountered during aggregation.

type StdLoggerAdapter ¶

type StdLoggerAdapter struct {
	ErrorLogger *log.Logger
	InfoLogger  *log.Logger
	DebugLogger *log.Logger
	TraceLogger *log.Logger
	// contains filtered or unexported fields
}

StdLoggerAdapter is a logger implementation, which sends all logs to provided standard output.

func (l *StdLoggerAdapter) Debug(msg string, fields LogFields)

func (l *StdLoggerAdapter) Error(msg string, err error, fields LogFields)

func (l *StdLoggerAdapter) Info(msg string, fields LogFields)

func (l *StdLoggerAdapter) Trace(msg string, fields LogFields)

func (l *StdLoggerAdapter) With(fields LogFields) LoggerAdapter