Wait Click Join in YTsaurus Flow (Go)

An example of a join pipeline that joins the streams of impressions (hit) and actions (action) by a shared key, using timers and an external state. This is a Go implementation of the similar C++ example.

Source code

Structure

The pipeline consists of three computations:

  • hit_reader and action_reader — native sources (TSwiftPassthroughOrderedSourceComputation) declared directly in the spec: they read two queues and publish the hit and action streams. They have no Go code. Each source declares a watermark_strategy, so the watermark of the pipeline is computed from the event time.
  • join (joinFunction) — a transform computation served by the companion.

The join computation works as follows:

  1. It receives messages from two input streams: hit and action.
  2. It accumulates the window of one impression in the external state /join-state.
  3. It sets a timer for the moment the waiting window closes.
  4. When the timer fires, it publishes the join result to the joined_action stream or simply clears the state.

Both input streams are grouped by hit_id and hit_time, so everything that belongs to one impression falls into one key. The length of the window is set by the wait_for_actions computation parameter from the spec.

main.go

The entry point: creating the pipeline, registering the only computation, and starting it.

Source code: main.go


type actionMessage struct {
	flow.YSONMessage
	HitID      string `yson:"hit_id"`
	HitTime    uint64 `yson:"hit_time"`
	IsClick    bool   `yson:"is_click"`
	ActionTime uint64 `yson:"action_time"`
}

type hitMessage struct {
	flow.YSONMessage
	HitID      string `yson:"hit_id"`
	HitTime    uint64 `yson:"hit_time"`
	HitPayload string `yson:"hit_payload"`
}

type joinedActionMessage struct {
	flow.YSONMessage
	HitID      string `yson:"hit_id"`
	HitTime    uint64 `yson:"hit_time"`
	IsClick    bool   `yson:"is_click"`
	ShowTime   uint64 `yson:"show_time"`
	ClickTime  uint64 `yson:"click_time"`
	HitPayload string `yson:"hit_payload"`
}

func main() {
	pipeline := flow.NewPipeline()
	pipeline.AddStreams(
		flow.NewYSONStream[actionMessage](actionStream),
		flow.NewYSONStream[hitMessage](hitStream),
		flow.NewYSONStream[joinedActionMessage](joinedStream),
	)
	pipeline.Add(flow.NewRowComputation("join", &joinFunction{}))

	if err := pipeline.Run(); err != nil {
		fmt.Fprintf(os.Stderr, "wait_click_join: %v\n", err)
		os.Exit(1)
	}
}

join_function.go

The main logic of the join: the function handles two streams and uses timers for window aggregation.

OnMessage

It drops late messages, appends its own field to the impression window, and sets the timer for closing the window:


type joinFunction struct{}

var (
	_ flow.RowFunction      = (*joinFunction)(nil)
	_ flow.RowTimerFunction = (*joinFunction)(nil)
)

func (*joinFunction) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	var hit *hitMessage
	var action *actionMessage
	var hitTime uint64
	switch msg.StreamID {
	case hitStream:
		hit = &hitMessage{}
		if err := msg.ConvertTo(hit); err != nil {
			return err
		}
		hitTime = hit.HitTime
	case actionStream:
		action = &actionMessage{}
		if err := msg.ConvertTo(action); err != nil {
			return err
		}
		hitTime = action.HitTime
	default:
		return fmt.Errorf("unhandled stream %q", msg.StreamID)
	}

	wait, err := waitForActions(rt)
	if err != nil {
		return err
	}
	closeTime := hitTime + wait
	if msg.EventTimestamp >= closeTime || msg.EventTimestamp < rt.MinWatermark() {
		return nil
	}

	state, err := flow.OpenExternalState(rt, joinStateName, msg)
	if err != nil {
		return err
	}
	var window joinState
	if _, err := state.ConvertTo(&window); err != nil {
		return err
	}

	if hit != nil {
		window.HitPayload = &hit.HitPayload
	} else if action.IsClick {
		window.ClickTime = &action.ActionTime
	} else {
		window.ShowTime = &action.ActionTime
	}
	if err := state.ConvertFrom(&window); err != nil {
		return err
	}

	out.AddTimer(flow.TimerRequest{TriggerTimestamp: closeTime, EventTimestamp: hitTime})
	return nil
}

OnTimer

It closes the window: it publishes the join result if both sides arrived, and clears the state in any case:


func (*joinFunction) OnTimer(
	ctx context.Context,
	rt flow.Runtime,
	timer flow.Timer,
	out flow.OutputCollector,
) error {
	if timer.StreamID != timerStream {
		return fmt.Errorf("unhandled timer stream %q", timer.StreamID)
	}

	state, err := flow.OpenExternalState(rt, joinStateName, timer)
	if err != nil {
		return err
	}
	var window joinState
	if _, err := state.ConvertTo(&window); err != nil {
		return err
	}

	if window.ShowTime != nil && *window.ShowTime != 0 && window.HitPayload != nil {
		var key joinKey
		if err := timer.Key.ConvertTo(&key); err != nil {
			return err
		}
		joined, err := joinedAction(rt, key, window)
		if err != nil {
			return err
		}
		out.AddMessage(joined)
	}

	return state.Clear()
}

Key patterns

  • External state through flow.OpenExternalState(rt, joinStateName, msg): ConvertTo reads the window into joinState, the message changes the field it needs in the structure, and ConvertFrom saves the window.
  • A timer with two timestamps: out.AddTimer(flow.TimerRequest{TriggerTimestamp: closeTime, EventTimestamp: hitTime}) — TriggerTimestamp sets when the window will close, EventTimestamp ties the timer to the event. Messages from both streams set the timer, not only hit: an action can arrive earlier than the impression it belongs to, and the window still has to be closed. A message that already falls outside its window or behind the watermark returns early and sets no timer.
  • Filtering out late data: msg.EventTimestamp < rt.MinWatermark() drops the messages of already closed windows — otherwise a second join result would be published for one impression. For details, see watermarks.
  • Computation parameters: rt.Parameters().Get(waitForActionsParameter, &spelled) reads the window length from the spec instead of hardcoding it.
  • Handling several input streams: the branch on msg.StreamID splits the logic for hit and action, and an unknown stream returns an error instead of a silent skip.
  • The key instead of the state: timer.Key carries hit_id and hit_time from group_by_schema, so the window holds only the data being joined — there is no need to duplicate the key in it.

See also

Previous
Retryable Async Request
Next
URL Downloader