batch

Documentation

Index

• Constants
• func ExportSourceByPKRange(ctx context.Context, db *sql.DB, cfg SourceConfig) ([]string, error)
• func ExportSourceFromRedis(ctx context.Context, connCfg RedisConnConfig, cfg RedisSourceConfig) ([]string, error)
• func ImportReduceOutputs(ctx context.Context, db *sql.DB, cfg SinkConfig) error
• func ImportReduceOutputsToRedis(ctx context.Context, connCfg RedisConnConfig, cfg RedisSinkConfig) error
• func OpenForApp(ctx context.Context, cfg DBConfig) (*sql.DB, error)
• func PrepareSyntheticSource(ctx context.Context, db *sql.DB, cfg PrepareConfig) error
• func RunFlow(ctx context.Context, cfg FlowConfig) error
• func RunMapReduce(ctx context.Context, cfg MapReduceRunConfig) error
• func RunPipeline(ctx context.Context, cfg PipelineConfig) error
• func SetDefaultRunner(r Runner)
• func ValidateAggregation(ctx context.Context, db *sql.DB, cfg ValidateConfig) error
• func ValidateFlowConfig(cfg FlowConfig) error
• type BenchmarkConfig
• type BenchmarkResult
  • func RunBenchmark(ctx context.Context, cfg BenchmarkConfig) (BenchmarkResult, error)
• type DBConfig
• type FlowBenchmarkResult
  • func RunFlowBenchmark(ctx context.Context, cfg FlowConfig) (FlowBenchmarkResult, error)
• type FlowConfig
• type FlowSinkConfig
• type FlowSourceConfig
• type FlowTransformConfig
• type LegacyRunner
  • func (LegacyRunner) Run(ctx context.Context, cfg MapReduceRunConfig) error
• type MapReduceRunConfig
• type PipelineConfig
• type PrepareConfig
• type RedisConnConfig
• type RedisSinkConfig
• type RedisSourceConfig
• type Runner
  • func DefaultRunner() Runner
• type SinkConfig
• type SourceConfig
• type ValidateConfig

Constants

const FlowVersionV1 = "v1"

Functions

func ExportSourceByPKRange

func ExportSourceByPKRange(ctx context.Context, db *sql.DB, cfg SourceConfig) ([]string, error)

func ExportSourceFromRedis

func ExportSourceFromRedis(ctx context.Context, connCfg RedisConnConfig, cfg RedisSourceConfig) ([]string, error)

func ImportReduceOutputs

func ImportReduceOutputs(ctx context.Context, db *sql.DB, cfg SinkConfig) error

func ImportReduceOutputsToRedis

func ImportReduceOutputsToRedis(ctx context.Context, connCfg RedisConnConfig, cfg RedisSinkConfig) error

func OpenForApp

func OpenForApp(ctx context.Context, cfg DBConfig) (*sql.DB, error)

OpenForApp opens a MySQL connection for advanced/custom flows.

func PrepareSyntheticSource

func PrepareSyntheticSource(ctx context.Context, db *sql.DB, cfg PrepareConfig) error

PrepareSyntheticSource creates a synthetic source table for benchmark.

func RunFlow

func RunFlow(ctx context.Context, cfg FlowConfig) error

RunFlow executes source -> transform -> sink defined by FlowConfig.

func RunMapReduce

func RunMapReduce(ctx context.Context, cfg MapReduceRunConfig) error

RunMapReduce executes map-reduce through the configured runner.

func RunPipeline

func RunPipeline(ctx context.Context, cfg PipelineConfig) error

RunPipeline executes MySQL source -> MapReduce -> MySQL sink in-process.

func SetDefaultRunner

func SetDefaultRunner(r Runner)

SetDefaultRunner overrides the process-wide runtime strategy.

func ValidateAggregation

func ValidateAggregation(ctx context.Context, db *sql.DB, cfg ValidateConfig) error

ValidateAggregation checks source(group by) equals target table data.

func ValidateFlowConfig

func ValidateFlowConfig(cfg FlowConfig) error

ValidateFlowConfig validates v1 flow schema and required fields.

Types

type BenchmarkConfig

type BenchmarkConfig struct {
	DB       DBConfig
	Prepare  bool
	PrepareC PrepareConfig
	Pipeline PipelineConfig
	Validate ValidateConfig
}

BenchmarkConfig configures benchmark workflow.

type BenchmarkResult

type BenchmarkResult struct {
	PrepareDuration  time.Duration
	PipelineDuration time.Duration
	ValidateDuration time.Duration
	TotalDuration    time.Duration
}

BenchmarkResult captures stage durations.

func RunBenchmark

func RunBenchmark(ctx context.Context, cfg BenchmarkConfig) (BenchmarkResult, error)

type DBConfig

type DBConfig struct {
	Host     string
	Port     int
	User     string
	Password string
	Database string
	Params   map[string]string
}

DBConfig defines MySQL connection parameters.

type FlowBenchmarkResult

type FlowBenchmarkResult struct {
	SourceDuration    time.Duration
	TransformDuration time.Duration
	SinkDuration      time.Duration
	TotalDuration     time.Duration
}

FlowBenchmarkResult captures source/transform/sink stage durations.

func RunFlowBenchmark

func RunFlowBenchmark(ctx context.Context, cfg FlowConfig) (FlowBenchmarkResult, error)

RunFlowBenchmark executes a config-driven flow and reports stage durations.

type FlowConfig

type FlowConfig struct {
	Version   string              `json:"version"`
	Source    FlowSourceConfig    `json:"source"`
	Transform FlowTransformConfig `json:"transform"`
	Sink      FlowSinkConfig      `json:"sink"`
}

FlowConfig describes a SeaTunnel-like source/transform/sink pipeline.

type FlowSinkConfig

type FlowSinkConfig struct {
	Type        string          `json:"type"`
	DB          DBConfig        `json:"db"`
	Redis       RedisConnConfig `json:"redis"`
	Config      SinkConfig      `json:"config"`
	RedisConfig RedisSinkConfig `json:"redis_config"`
}

type FlowSourceConfig

type FlowSourceConfig struct {
	Type        string            `json:"type"`
	DB          DBConfig          `json:"db"`
	Redis       RedisConnConfig   `json:"redis"`
	Config      SourceConfig      `json:"config"`
	RedisConfig RedisSourceConfig `json:"redis_config"`
}

type FlowTransformConfig

type FlowTransformConfig struct {
	Type       string            `json:"type"`
	Builtin    string            `json:"builtin"`
	PluginPath string            `json:"plugin_path"`
	Reducers   int               `json:"reducers"`
	Workers    int               `json:"workers"`
	InRAM      bool              `json:"in_ram"`
	Port       int               `json:"port"`
	Params     map[string]string `json:"params"`
}

type LegacyRunner

type LegacyRunner struct{}

LegacyRunner uses the current in-process legacy mapreduce runtime.

func (LegacyRunner) Run

func (LegacyRunner) Run(ctx context.Context, cfg MapReduceRunConfig) error

type MapReduceRunConfig

type MapReduceRunConfig struct {
	Files      []string
	PluginPath string
	Reducers   int
	Workers    int
	InRAM      bool
	Port       int
}

MapReduceRunConfig describes a runtime invocation for a map-reduce job.

type PipelineConfig

type PipelineConfig struct {
	DB         DBConfig // backward compatibility fallback when SourceDB/SinkDB are not set
	SourceDB   DBConfig
	SinkDB     DBConfig
	Source     SourceConfig
	Sink       SinkConfig
	PluginPath string
	Reducers   int
	Workers    int
	InRAM      bool
	Port       int
}

PipelineConfig describes end-to-end MySQL -> MapReduce -> MySQL job.

type PrepareConfig

type PrepareConfig struct {
	SourceTable string
	Rows        int64
	KeyMod      int64
}

PrepareConfig configures synthetic source table generation for benchmarking.

type RedisConnConfig

type RedisConnConfig = redis_batch.ConnConfig

type RedisSinkConfig

type RedisSinkConfig = redis_batch.SinkConfig

type RedisSourceConfig

type RedisSourceConfig = redis_batch.SourceConfig

type Runner

type Runner interface {
	Run(ctx context.Context, cfg MapReduceRunConfig) error
}

Runner abstracts runtime startup strategy for map-reduce execution.

func DefaultRunner

func DefaultRunner() Runner

DefaultRunner returns the current process-wide runtime strategy.

type SinkConfig

type SinkConfig = mysql_batch.SinkConfig

type SourceConfig

type SourceConfig = mysql_batch.SourceConfig

Unified source/sink config aliases exposed by batch package.

type ValidateConfig

type ValidateConfig struct {
	SourceTable string
	SourceKey   string
	SourceVal   string
	TargetTable string
	TargetKey   string
	TargetVal   string
}

ValidateConfig compares source aggregation with target table.