問題タブ [netflow]

私は巨大な NetFlow データベースを持っています (タイムスタンプ、送信元 IP、送信先 IP、プロトコル、送信元および送信先ポート番号、交換されたパケット数、バイト数などが含まれています)。現在および前の行に基づいてカスタム属性を作成したいと考えています。

現在の行のソース IP とタイムスタンプに基づいて新しい列を計算したいと考えています。これは私が論理的にやりたいことです:

• 現在の行のソース IP を取得します。
• 現在の行のタイムスタンプを取得します。
• ソース IP とタイムスタンプに基づいて、ソース IP と一致するデータフレーム全体のすべての前の行を取得したいのですが、通信は過去 30 分間に発生しました。これはとても重要です。
• 基準 (ソース IP と過去 30 分間に発生したもの) に一致する行 (私の例ではフロー) について、すべてのパケットとすべてのバイトの合計と平均を数えたいと思います。

データセットからの 1 行

関連するコードのスニペット:

これを行うには、おそらくもっと効率的で優れた方法があることはわかっていますが、残念ながら私は最高のプログラマーではありません。

ありがとう。

ClickHouse テーブル (実際には、netflow コレクターからの netflow V9) に生の JSON データがいくつかあります。次のようになります。

私の仕事は、次の方法で DataSets 配列を別の ClickHouse テーブルに変換することです。