flume - Flume バッチサイズ

Question

私のflume設定は

agent.sinks.sink.batchSize=10

どのフルームプロセス10 reacords at each batchでも、すべてのレコードで同じ時間が得られます。

そのため、バッチ内の各レコードの個別の時間を取得することができます。

Answer 1

その他のオプションは、生成時にタイムスタンプ ヘッダーを設定するか、文字列を解析してそのようにタイムスタンプを設定することにより、イベントのタイムスタンプを正しく取得することです。

Answer 2

タイムスタンプの設定にタイムスタンプ インターセプタを使用していますか?

そのソースコードを見てみましょう:

  /**
   * Modifies events in-place.
   */
  @Override
  public Event intercept(Event event) {
    Map<String, String> headers = event.getHeaders();
    if (preserveExisting && headers.containsKey(TIMESTAMP)) {
      // we must preserve the existing timestamp
    } else {
      long now = System.currentTimeMillis();
      headers.put(TIMESTAMP, Long.toString(now));
    }
    return event;
  }

  /**
   * Delegates to {@link #intercept(Event)} in a loop.
   * @param events
   * @return
   */
  @Override
  public List<Event> intercept(List<Event> events) {
    for (Event event : events) {
      intercept(event);
    }
    return events;
  }

メソッドの実行時間intercept(List<Event> events)は非常に短いため、バッチ内のすべてのイベントが 1 ミリ秒以内に処理されると思います。そのため、このインターセプターを使用してバッチ内のイベントに異なるタイムスタンプを強制することはできません。

バッチ内の各イベントに一意の識別子が必要な場合は、タイムスタンプにバッチ内のイベント数を追加する TimestampInterceptor コードに基づいて独自のインターセプターを作成できます。ただし、2 つのバッチが 1 ミリ秒以内に処理される可能性があるため、インテントファイアのグローバルな一意性は保証されません。

より具体的なアドバイスが必要な場合は、要件を明確にしてください。