worker

Documentation

Index

• Variables
• type LocalCache
  • func LoadLocalCache(path string) (*LocalCache, error)
  • func (c *LocalCache) WriteLocalCache() error
• type Worker
  • func NewWorker(masterHost string, masterPort int, httpHost string, httpPort, jobs int, ...) *Worker
  • func (w *Worker) Pull()
  • func (w *Worker) Recommend(m ranking.MatrixFactorization, users []string)
  • func (w *Worker) Serve()
  • func (w *Worker) ServeMetrics()
  • func (w *Worker) Sync()

Variables

var (
	GenerateRecommendSeconds = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_get_recommend_seconds",
	})
	CTRRecommendSeconds = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_ctr_recommend_seconds",
	})
	CollaborativeRecommendSeconds = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_collaborative_recommend_seconds",
	})
	ItemBasedRecommendSeconds = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_item_based_recommend_seconds",
	})
	UserBasedRecommendSeconds = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_user_based_recommend_seconds",
	})
	LoadLatestRecommendCacheSeconds = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_load_latest_recommend_cache_seconds",
	})
	LoadPopularRecommendCacheSeconds = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_load_popular_recommend_cache_seconds",
	})

	GenerateRecommendTimes = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_get_recommend_times",
	})
	CTRRecommendTimes = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_ctr_recommend_times",
	})
	CollaborativeRecommendTimes = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_collaborative_recommend_times",
	})
	ItemBasedRecommendTimes = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_item_based_recommend_times",
	})
	UserBasedRecommendTimes = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_user_based_recommend_times",
	})
	LoadLatestRecommendCacheTimes = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_load_latest_recommend_cache_times",
	})
	LoadPopularRecommendCacheTimes = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "gorse",
		Subsystem: "worker",
		Name:      "worker_load_popular_recommend_cache_times",
	})
)

Types

type LocalCache

type LocalCache struct {
	WorkerName string
	// contains filtered or unexported fields
}

LocalCache for the worker node.

func LoadLocalCache

func LoadLocalCache(path string) (*LocalCache, error)

LoadLocalCache loads cache from a local file.

func (*LocalCache) WriteLocalCache

func (c *LocalCache) WriteLocalCache() error

WriteLocalCache writes cache to a local file.

type Worker

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

Worker manages states of a worker node.

func NewWorker

func NewWorker(masterHost string, masterPort int, httpHost string, httpPort, jobs int, cacheFile string) *Worker

NewWorker creates a new worker node.

func (*Worker) Pull

func (w *Worker) Pull()

Pull user index and ranking model from master.

func (*Worker) Recommend

func (w *Worker) Recommend(m ranking.MatrixFactorization, users []string)

Recommend items to users. The workflow of recommendation is: 1. Skip inactive users. 2. Load historical items. 3. Load positive items if KNN used. 4. Generate recommendation. 5. Save result. 6. Insert cold-start items into results. 7. Rank items in results by click-through-rate. 8. Refresh cache.

func (*Worker) Serve

func (w *Worker) Serve()

Serve as a worker node.

func (*Worker) ServeMetrics

func (w *Worker) ServeMetrics()

ServeMetrics serves Prometheus metrics.

func (*Worker) Sync

func (w *Worker) Sync()

Sync this worker to the master.