(57)【特許請求の範囲】
【請求項１】
プロセッサとメモリを有する計算機でタイムスタンプが付与されたタプルを処理するストリームデータの分散処理方法であって、
前記計算機が、入力されたタプルを処理するクエリと、当該クエリが出力するタプルに遅延時間を加算する可変遅延演算と、前記可変遅延演算の出力を前記クエリに再帰的に入力する再帰ループを含むクエリ定義を受け付ける第１のステップと、
前記計算機が、前記タプルの入力先と、前記タプルの出力先と、遅延時間を含む構成変更命令を受け付ける第２のステップと、
前記計算機が、前記再帰ループを解除して前記可変遅延演算の出力を前記構成変更命令の出力先に変更し、前記タプルの入力先を前記構成変更命令の入力先に変更し、前記構成変更命令の遅延時間を前記可変遅延演算に設定する第３のステップと、
前記計算機が、前記入力先からタプルを受け付けて、前記クエリで所定の処理を行ったタプルを前記可変遅延演算に入力し、当該タプルのタイムスタンプに前記遅延時間を加算する第４のステップと、
前記計算機が、前記出力先へ前記遅延時間を加算されたタプルを出力する第５のステップと、
を含むことを特徴とするストリームデータの分散処理方法。
【請求項２】
請求項１に記載のストリームデータの分散処理方法であって、
前記第２のステップは、
前記出力先と前記遅延時間の関係を出力先別遅延時間管理情報に格納し、
前記第３のステップは、
前記出力先毎に遅延時間を格納した出力先別遅延時間管理情報から出力先に応じた遅延時間を前記タイムスタンプに加算することを特徴とするストリームデータの分散処理方法。
【請求項３】
請求項１に記載のストリームデータの分散処理方法であって、
前記構成変更命令は、当該構成変更命令の実行開始時刻を含み、
前記第２のステップは、
前記構成変更命令を、構成変更命令キューに格納し、
前記第３のステップは、
入力された前記タプルのタイムスタンプが前記構成変更命令キューの先頭の構成変更命令の前記実行開始時刻を経過していれば、前記入力先と前記出力先及び前記遅延時間の設定を実施することを特徴とするストリームデータの分散処理方法。
【請求項４】
請求項１に記載のストリームデータの分散処理方法であって、
前記クエリ定義は、前記可変遅延演算の遅延時間を変数で定義したことを特徴とするストリームデータの分散処理方法。
【請求項５】
プロセッサとメモリを有する計算機と、
複数の前記計算機を管理する管理計算機と、
タイムスタンプが付与されたタプルを前記複数の計算機へ供給するデータソースと、を含むストリームデータの分散処理システムであって、
前記管理計算機は、
入力されたタプルを処理するクエリと、当該クエリが出力するタプルに遅延時間を加算する可変遅延演算と、前記可変遅延演算の出力を前記クエリに再帰的に入力する再帰ループを含むクエリ定義を前記複数の計算機へ分配するクエリ分配部と、
前記タプルの入力先と、前記タプルの出力先と、遅延時間を含む構成変更命令を前記複数の計算機へ分配する構成変更命令分配部と、を有し、
前記計算機は、
前記クエリ定義を受け付けてから、前記構成変更命令を受け付けると、前記再帰ループを解除して前記可変遅延演算の出力を前記構成変更命令の出力先に変更し、前記タプルの入力先を前記構成変更命令の入力先に変更するクエリ処理部と、
前記構成変更命令の遅延時間を加算する可変遅延演算実行部と、を有し、
前記クエリ処理部が、前記入力先からタプルを受け付けて、前記クエリで所定の処理を行ったタプルを前記可変遅延演算に入力し、
前記可変遅延演算実行部が、前記タプルのタイムスタンプに前記遅延時間を加算して、前記出力先へ出力することを特徴とするストリームデータの分散処理システム。
【請求項６】
請求項５に記載のストリームデータの分散処理システムであって、
前記計算機は、
前記出力先と前記遅延時間の関係を出力先別遅延時間管理情報に格納し、
前記可変遅延演算実行部は、
前記出力先毎に遅延時間を格納した出力先別遅延時間管理情報から出力先に応じた遅延時間を前記タイムスタンプに加算することを特徴とするストリームデータの分散処理システム。
【請求項７】
請求項５に記載のストリームデータの分散処理システムであって、
前記構成変更命令は、当該構成変更命令の実行開始時刻を含み、
前記計算機は、
前記構成変更命令を、構成変更命令キューに格納し、
前記クエリ処理部は、
入力された前記タプルのタイムスタンプが前記構成変更命令キューの先頭の構成変更命令の実行開始時刻を経過していれば、前記入力先と前記出力先及び前記遅延時間の設定を実施することを特徴とするストリームデータの分散処理システム。
【請求項８】
請求項５に記載のストリームデータの分散処理システムであって、
前記クエリ定義は、前記可変遅延演算の遅延時間を変数で定義したことを特徴とするストリームデータの分散処理システム。
【請求項９】
プロセッサとメモリを有する計算機を制御するプログラムを格納した記憶媒体であって、
入力されたタプルを処理するクエリと、当該クエリが出力するタプルに遅延時間を加算する可変遅延演算と、前記可変遅延演算の出力を前記クエリに再帰的に入力する再帰ループを含むクエリ定義を受け付ける第１のステップと、
前記タプルの入力先と、前記タプルの出力先と、遅延時間を含む構成変更命令を受け付ける第２のステップと、
前記再帰ループを解除して前記可変遅延演算の出力を前記構成変更命令の出力先に変更し、前記タプルの入力先を前記構成変更命令の入力先に変更し、前記構成変更命令の遅延時間を前記可変遅延演算に設定する第３のステップと、
前記入力先からタプルを受け付けて、前記クエリで所定の処理を行ったタプルを前記可変遅延演算に入力し、当該タプルのタイムスタンプに前記遅延時間を加算する第４のステップと、
前記出力先へ前記遅延時間を加算されたタプルを出力する第５のステップと、
を前記計算機に実行させるプログラムを格納した非一時的な計算機読み取り可能な記憶媒体。
【請求項１０】
請求項９に記載の記憶媒体であって、
前記第２のステップは、
前記出力先と前記遅延時間の関係を出力先別遅延時間管理情報に格納し、
前記第３のステップは、
前記出力先毎に遅延時間を格納した出力先別遅延時間管理情報から出力先に応じた遅延時間を前記タイムスタンプに加算することを特徴とする記憶媒体。
【請求項１１】
請求項９に記載の記憶媒体であって、
前記構成変更命令は、当該構成変更命令の実行開始時刻を含み、
前記第２のステップは、
前記構成変更命令を、構成変更命令キューに格納し、
前記第３のステップは、
入力された前記タプルのタイムスタンプが前記構成変更命令キューの先頭の構成変更命令の実行開始時刻を経過していれば、前記入力先と前記出力先及び前記遅延時間の設定を実施することを特徴とする記憶媒体。
【請求項１２】
請求項９に記載の記憶媒体であって、
前記クエリ定義は、前記可変遅延演算の遅延時間を変数で定義したことを特徴とする記憶媒体。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、ストリームデータ処理の分散処理方法に関する。
【背景技術】
【０００２】
株取引の自動化や、交通情報処理の高度化や、ネットワークのマイクロバーストの検出や、ＩｏＴ（Internet of Things）のセンサデータや、ソーシャルネットワークの解析といった、高いレートで継続的に発生する情報をリアルタイムで解析し、重要なイベントの発生に対して瞬時にアクションを起こす要求の高まりを背景に、高レートデータのリアルタイム処理を実現する、ストリームデータ処理（以下、ストリーム処理）が注目されている。【０００３】
ストリーム処理は、様々なデータ処理に適用可能な汎用ミドルウェア技術であるため、個別案件ごとにシステムを構築するのでは間に合わないようなビジネス環境の急激な変化にも応えつつ、実世界のデータをリアルタイムにビジネスに反映することを可能とする。【０００４】
ストリーム処理が対象とするストリームとは、タイムスタンプ付きのデータであるタプルが、連続（または断続的）して到来する時系列データである。ストリーム処理のユーザが、このストリームに対する監視ルールをクエリとして定義すると、ストリーム処理サーバのクエリ登録インタフェースが、クエリ定義をクエリグラフに変換する。【０００５】
クエリグラフは、オペレータと呼ばれる処理単位をノードとし、同オペレータ間のタプルキューをエッジとする、有向グラフである。ストリーム処理では、入力ストリームを構成する個々のタプルについて、クエリグラフを通過させることで、データフロー的に処理を進める。【０００６】
ストリーム処理は、データフロー型の処理であるため、クエリグラフを多段に分割し、複数の計算資源によってパイプライン的に分散処理することが可能である。また、株価の銘柄別のチャート分析のように、互いに独立した多数の系列の時系列データを集計するクエリについては、データを系列別に分割することで、並列的に分散処理することが可能である。このように、ストリーム処理では、クエリの処理に応じて適切に分散処理することで、スループットの向上に繋がる。【０００７】
また、ＩｏＴのセンサデータの分析に基づく工場の製造機器監視や制御、あるいは自動車の車両間通信に基づく衝突回避などのような、機械制御へのストリーム処理の適用においては、タプルのレイテンシの短縮が重要となる。特に、個別の製造機器にとどまらない、製造工程全体を俯瞰したセンサデータの分析、あるいは自動車の衝突回避のための複数車両による協調処理を実現するためには、複数のストリーム処理が連携する分散処理と、レイテンシの短縮とを両立する必要がある。【０００８】
ストリーム処理の従来技術として、特許文献１が知られている。特許文献１には、ストリームデータ処理システムのストリームデータのクエリ処理に関するスループットを向上させる方法が開示されており、「本願明細書に記載されている技術は、データストリーム処理タスクの宣言的な記述から、高度に最適化された分散ストリーム処理アプリケーションを作成する、コード生成フレームワークを含む。本発明の一つ以上の実施例によれば、開発者は、分散システムに存在しがちな性能影響を気にすることなく、細粒度のストリーム演算子を用いて、アプリケーションを構築することができる。さらに、本願明細書に記載されている技術は、アプリケーション開発において、粗粒度の演算子を使用した場合に生じるような非柔軟性を回避する。本願明細書に記載されている典型的なコード生成フレームワークは、通信オーバヘッドを最小化するように、アプリケーションを適切なサイズの実行ユニットへ割り当てることができる。」と記載されている。【０００９】
また、ストリーム処理の従来技術として特許文献２が知られている。特許文献２には、ストリームデータ処理システムのストリームデータのクエリ処理に関するレイテンシを短縮する方法が開示されており、「ストリームデータ処理システムは、複数のクエリの実行木から単一のオペレータグラフを構築し、ストリーム化演算の実行が、入力から出力に向かって一方通行で進むように、オペレータの実行順序を決定し、システム外部のデータを入力する外部発火オペレータと時限的にデータを生成する内部発火オペレータの発火時刻を監視し、オペレータ実行制御部は、最も早い発火時刻のオペレータを起点として、決定したオペレータ実行順序に従って、当該時刻のオペレータグラフにおける処理を完結させる処理を繰り返す。」と記載されている。【００１０】
また、特許文献２には、遅延演算を用いた再帰クエリの実行方法が開示されており、「ストリームタプルのタイムスタンプを未来にずらす、遅延オペレータである遅延演算を挿入した様子を示している。時刻の遅延が挟まることで、ストリーム化演算の出力は、自分自身の入力に時間がずれて戻ることになるため、時刻ｔ１におけるリレーションの増減に対し、該ストリーム化演算自身の出力が影響を持つことは無くなる。このように、再帰クエリの途中に遅延演算を挟むことによって、ストリーム化演算のデッドロックを回避し、再帰クエリを実現することが可能となる。」と記載されている。
【先行技術文献】
【特許文献】
【００１１】
【特許文献１】米国特許出願公開第２０１２／０２５９９１０号明細書
【特許文献２】特開２０１４−６７４５６号公報
【発明の概要】
【発明が解決しようとする課題】
【００１２】
しかしながら、ストリーム処理は、スライディングウィンドウのように、留まることのないデータの流れから過去一定の範囲のデータセットを継続して切り出し、宣言的なクエリ言語による連続処理を可能とする。すなわち、時間軸方向の変化に対する処理を暗黙に含んでおり、複雑な制御を要するリアルタイムアプリケーションの開発工数を大幅に削減する。【００１３】
ただし、このようなリアルタイムアプリケーションの開発においては、デバグのために、処理の再現性（すなわち、タイムスタンプまで含めて同一のタプル系列は、何回処理しても、その結果ストリームのタプル系列が不変であること）が確保されることが必須である。【００１４】
処理の再現性を維持するために、分散処理では、互いに異なる計算機上で稼働する複数のストリーム処理プロセスが生成する結果ストリームを、時刻順を維持しながら、マージして処理を伝搬させていく必要がある。このようなマージ処理は、複数のストリーム処理プロセスから到来する結果ストリームの待ち合わせを伴うため、結果ストリームを多段にマージするケースなどでは、ストリーム処理の時間やプロセス間通信時間の累積に起因するレイテンシの増加が発生する。【００１５】
また、複数のストリーム処理プロセス間の入出力関係が循環を含むようなケースでは、両プロセス間の相互依存関係により、クエリ処理の進行がハングアップする場合もある。例えば、自動車の衝突回避のような協調処理においては、入出力の循環を必然的に含むことになる。【００１６】
さらに、従来のストリーム処理の分散実行技術では、ストリーム処理プロセス間の入出力関係に合せて、各プロセス上で稼働するクエリ同士の入出力関係が定まっている必要がある。従って、ユーザは各クエリを実行するプロセスの実行位置（実行する計算機）を意識して、クエリ同士の接続関係、即ちクエリグラフを設計する必要がある。クエリ同士の接続関係の制約は、ユーザによる分散クエリ定義の容易性を著しく悪化させる。【００１７】
また、一旦分散実行を開始したクエリグラフについて、クエリ間の接続関係の一部を変更する場合でも、クエリグラフ全体を停止して、クエリグラフを定義し直して、同クエリグラフの処理を再開する必要がある。従って、その間の分析処理の停止を伴う。【００１８】
特許文献１の技術では、クエリグラフの定義の中に、各クエリの実行位置の指定を含めることができる。但し、実行位置を意識したクエリグラフ定義、およびクエリグラフの変更に伴う処理の停止及び再起動が必要である、という問題があった。【００１９】
特許文献２の技術では、単一計算機上の単一プロセスの単一スレッドにおいて、クエリグラフの処理を低レイテンシで実行し、複数のクエリ間の循環、即ち再帰クエリを実行することができる。但し、単一スレッド実行が前提であるため、複数の計算機に跨るクエリグラフの分散実行においては、同様の効果を得ることができない、という問題があった。【００２０】
本発明は上記問題点に鑑みて成されたものであり、ストリーム処理の分散実行における、クエリ定義の容易化と、処理の停止を伴わずに動的なクエリ間の接続関係の変更を可能にすることを目的とする。
【課題を解決するための手段】
【００２１】
プロセッサとメモリを有する計算機でタイムスタンプが付与されたタプルを処理するストリームデータの分散処理方法であって、前記計算機が、入力されたタプルを処理するクエリと、当該クエリが出力するタプルに遅延時間を加算する可変遅延演算と、前記可変遅延演算の出力を前記クエリに再帰的に入力する再帰ループを含むクエリ定義を受け付ける第１のステップと、前記計算機が、前記タプルの入力先と、前記タプルの出力先と、遅延時間を含む構成変更命令を受け付ける第２のステップと、前記計算機が、前記再帰ループを解除して前記可変遅延演算の出力を前記構成変更命令の出力先に変更し、前記タプルの入力先を前記構成変更命令の入力先に変更し、前記構成変更命令の遅延時間を前記可変遅延演算に設定する第３のステップと、前記計算機が、前記入力先からタプルを受け付けて、前記クエリで所定の処理を行ったタプルを前記可変遅延演算に入力し、当該タプルのタイムスタンプに前記遅延時間を加算する第４のステップと、前記計算機が、前記出力先へ前記遅延時間を加算されたタプルを出力する第５のステップと、を含む。
【発明の効果】
【００２２】
上記本発明によれば、ストリーム処理の再現性を維持しながら、クエリ定義の容易化と、処理停止を伴わない動的なクエリ間の接続関係の変更が可能となる。
【図面の簡単な説明】
【００２３】
【図１】本発明の実施例１を示し、ストリーム処理サーバが稼働する計算機システムの一例を示すブロック図である。
【図２】本発明の実施例１を示し、計算機システムを工場の生産ラインの監視処理に適用した例を示すブロック図である。
【図３】本発明の実施例１を示し、計算機システムを自動車の車両間通信に適用した例を示すブロック図である。
【図４】本発明の従来例を示し、工場の生産ラインを監視する継続クエリの一例を示すスクリプトである。
【図５】従来例を示し、工場の生産ラインを監視する、継続クエリのクエリグラフの一例を示すブロック図である。
【図６】従来例を示し、工場の生産ラインを監視する、ストリーム処理プロセス間連携の一例を示すブロック図である。
【図７】従来例を示し、自動車の車両間通信を実現する、ストリーム処理プロセス間連携の一例を示すブロック図である。
【図８】本発明の実施例１を示し、可変遅延演算を挿入した再帰クエリを工場の生産ラインの監視処理に適用した例を示すブロック図である。
【図９Ａ】本発明の実施例１を示し、可変遅延演算を挿入した再帰クエリの一例を示すスクリプトである。
【図９Ｂ】本発明の実施例１を示し、可変遅延演算を挿入した再帰クエリの一例を示すスクリプトである。
【図１０】本発明の実施例１を示し、ストリーム処理プロセスの構成を示すブロック図である。
【図１１】本発明の実施例１を示し、構成変更命令の一例を示すブロック図である。
【図１２Ａ】本発明の実施例１を示し、結果タプルキューの一例を示す図である。
【図１２Ｂ】本発明の実施例１を示し、構成変更命令キューの一例を示す図である。
【図１２Ｃ】本発明の実施例１を示し、出力先別遅延時間管理表の一例を示す図である。
【図１２Ｄ】本発明の実施例１を示し、可変遅延演算実行部の一例を示す図である。
【図１３】本発明の実施例１を示し、可変遅延演算実行部で行われる処理の一例を示すフローチャートである。
【図１４】本発明の実施例１を示し、工場の生産ラインを監視する、ストリーム処理プロセス間の、遅延演算を介した連携の一例を示すブロック図である。
【図１５】本発明の実施例１を示し、自動車の車両間通信を実現する、ストリーム処理プロセス間の、遅延演算を介した連携の一例を示すブロック図である。
【図１６】本発明の実施例１を示し、自動車の車両間通信を実現する、ストリーム処理プロセス間の、遅延演算を介した連携の一例を示すブロック図である。
【図１７】本発明の第２の実施例を示し、ストリーム処理プロセスの構成を示すブロック図である。
【発明を実施するための形態】
【実施例１】
【００２４】
以下、本発明の一実施形態について添付図面を用いて説明する。【００２５】
図１は、ストリーム処理サーバが稼働する計算機システム（ストリーム処理分散実行システム８０１）の一例を示すブロック図である。計算機システムは、複数のストリーム処理サーバ１００−１〜１００−ｎと、ストリーム処理サーバ１００−１〜１００−ｎにクエリ定義１３２を分配するストリーム処理分散実行管理サーバ１００１と、クエリ定義１３２や構成変更命令８１５を発行するホスト計算機１３０と、タプル１２１を供給するデータソース１２０と、結果タプル１４１を受信するデータ受信機１４０と、各計算機を接続するネットワーク１５０と、を含む。【００２６】
ストリーム処理サーバ１００−１は、ＣＰＵ１０１と、メモリ１０３と、ネットワークインタフェース１０５と、ストレージ１０６、およびそれらを結合するバス１０４によって構成される計算機である。なお、ストリーム処理サーバ１００−１〜１００−ｎの構成であるので、重複する説明は省略する。【００２７】
メモリ１０３上には、ストリーム処理の論理的な手順を定義するストリーム処理プロセス１１０−１が配置される。ストリーム処理プロセス１１０−１は、ＣＰＵ１０１によって実行可能な実行イメージ（プログラム等）である。【００２８】
ストリーム処理サーバ１００−１は、ネットワークインタフェース１０５を介してネットワーク１５０に接続される。ストリーム処理サーバ１００−１は、ネットワーク１５０に接続されたストリーム処理分散実行管理サーバ１００１からクエリ定義１３２と構成変更命令８１５を受け付ける。【００２９】
ネットワーク１５０に接続されたホスト計算機１３０上では、クエリ登録インタフェース１３１によって、ユーザが指定したクエリ定義１３２を、ストリーム処理分散実行管理サーバ１００１へ出力する。【００３０】
ホスト計算機１３０は、ＣＰＵ１２とメモリ１３及び表示装置とキーボードなどの入力装置を含む入出力装置１３３を有し、メモリ１３にクエリ登録インタフェース１３１をロードして実行し、クエリ定義１３２や構成変更命令８１５等の編集やストリーム処理プロセスの割当てなどを行う。【００３１】
次に、ストリーム処理プロセス１１０−１は、上記クエリ定義１３２に従ってストリーム処理を実行可能なクエリグラフを構築する。この後、ネットワーク１５０に接続されたデータソース１２０から送信されるタプル１２１を、ストリーム処理サーバ１００−１が受け取ると、ストリーム処理サーバ１００−１は該クエリグラフに従って時系列のタプル１２１をクエリで処理し、結果タプル１４１を生成する。【００３２】
この結果タプル１４１は、ネットワーク１５０に接続されたデータ受信機１４０に送信される。ストレージ１０６は、ストリーム処理プロセス１１０−１の実行イメージの他、一度受け取った前記クエリ定義１３２のテキストファイルを保存する。ストリーム処理プロセス１１０−１は、起動時にストレージ１０６からこのクエリ定義１３２のファイルをロードし、クエリグラフを構築することも可能である。【００３３】
ネットワーク１５０には、複数のストリーム処理サーバ１００−１〜１００−ｎが接続される。ストリーム処理サーバ１００−２〜１００−ｎは、それぞれストリーム処理サーバ１００−１と同様の構成要素を有し、それぞれが有するメモリ１０３上に、ストリーム処理プロセス１１０−１の実行イメージを保持すする。【００３４】
なお、以下の説明ではストリーム処理サーバのそれぞれを特定する際には"−"を含む符号１００−１〜１００−ｎを用い、ストリーム処理サーバの全体については"−"のない符号１００で示す。他の構成要素の符号についても同様である。【００３５】
図２は、計算機システムを工場の生産ラインの監視処理に適用した例を示すブロック図である。ストリーム処理サーバ１０１−１〜１０１−５は、製造機器に付帯して稼働する計算機であり、それぞれストリーム処理プロセス１１０−１〜１１０−５を実行する。ストリーム処理サーバ１０１−１〜１０１−４には、それぞれセンサ２００−１〜２００−４が接続されており、各センサが生成するセンサデータは、それぞれストリーム処理プロセス１１０−１〜１１０−４で分析される。【００３６】
センサ２００−１と２００−４は、ベルトコンベア２１０上を流れる製造物２２０、２２１のデータを取得しており、センサ２００−２と２００−３は、ベルトコンベア２１１上を流れる製造物２２２、２２３、２２４のデータを取得している。【００３７】
製造物２２０は、まずセンサ２００−１でデータを取得されてから、数秒後に、センサ２００−４でデータを取得される。製造物２２１についても同様である。また、製造物２２２は、まずセンサ２００−２でデータを取得されてから、数秒後に、センサ２００−３でデータを取得される。製造物２２３、２２４についても同様である。【００３８】
ストリーム処理プロセス１１０−１〜１１０−５は、それぞれセンサデータの平均値、最大値、最小値などの、データ分析値を算出し、相互に連携して、生産ライン全体の分析値を集計する。【００３９】
生産ラインの上流に位置するストリーム処理プロセス１１０−１、１１０−２から、下流に位置するストリーム処理プロセス１１０−５に向かって、分析値を順次集約する。また、各ストリーム処理プロセス１１０は、不良品の発生に繋がるような異常データの発生を監視する。複数の製造機器を跨ったデータ分析の必要性を鑑みて、ストリーム処理プロセス１１０同士の接続関係は、生産ラインの流れに沿って定義される。製造物が変更されて、生産ラインの流れや、異常データの発生を検出する分析手法などが変更になる場合は、ストリーム処理プロセス１１０同士の接続関係も変更される。この接続関係の変更は、後述する構成変更命令８１５で行うことができる。【００４０】
図３は、図１の計算機システムを自動車の車両間通信に適用した例を示すブロック図である。各車両３００−１〜３００−４が有するストリーム処理サーバ１０１−１〜１０１−４は、例えば、車載マイクロコンピュータであり、それぞれストリーム処理プロセス１１０−１〜１１０−４を実行する。ストリーム処理プロセス同士は相互に連携し、ＧＰＳ２３０や各車両３００が有する各種センサから位置情報などのデータを取得し、互いの位置情報や、速度などの情報を交換し合って、車両３００毎の走行予定の経路を決定する。車両同士の接続関係は、相互の位置関係に従って随時変更される。【００４１】
図４は、従来例を示し、工場の生産ラインを監視する、継続クエリの一例を示すスクリプトである。生産ライン全体の分析を定義するクエリ４００を示しており、冒頭は、センサ２００−１〜２００−４が取得するデータを入力ストリームとして定義する、図中Ｓ１〜Ｓ４のｒｅｇｉｓｔｅｒ  ｓｔｒｅａｍ句が列挙されている。【００４２】
クエリＱ３（ｑｕｅｒｙ  Ｑ３）は、ストリーム処理プロセス１１０−３で実行される分析クエリを表しており、ｆｒｏｍ句によって、ストリーム処理プロセス１１０−２で稼働する分析クエリＱ２（ｑｕｅｒｙ  Ｑ２）の結果と、センサ２００−３の入力ストリームＳ３を突き合わせて、分析することを意味している。【００４３】
クエリＱ１２（ｑｕｅｒｙ  Ｑ１２）は、ストリーム処理プロセス１１０−１で稼働する分析クエリＱ１の結果と、分析クエリＱ２の結果をマージする。クエリＱ４は、ストリーム処理プロセス１１０−４で稼働する分析クエリを表しており、ｆｒｏｍ句によって、クエリＱ１２の出力と、センサ２００−４の入力ストリームＳ４を突き合わせて、分析することを意味している。【００４４】
クエリの途中に挿入されているアノテーション「ｅｘｅｃ＿ｏｎ」は、各クエリを実行するストリーム処理サーバ（および、そこで稼働するストリーム処理プロセス）の識別子を定義する、実行ノード指定アノテーションである。クエリＱ３はストリーム処理プロセス１１０−３で、クエリＱ１２とＱ４はストリーム処理プロセス１１０−４で、それぞれ実行されることを意味している。【００４５】
本実施例１では記載を省略しているが、他のストリーム処理プロセス１１０−１、１１０−２、１１０−５で稼働するクエリの定義についても同様に定義される。【００４６】
図５は、工場の生産ラインを監視する、継続クエリのクエリグラフの一例を示すブロック図である。図４に示したクエリ４００中の、各ストリーム処理プロセスで稼働する分析クエリの定義に従って、入力ストリーム５０１〜５０３や分析クエリ５１１〜５１５間の入出力関係が定まる。【００４７】
入力ストリーム５０１〜５０４は、図４のクエリ４００のストリームＳ１〜Ｓ４に対応する。分析クエリ５１１〜５１５は、クエリ４００のアノテーションによって、それぞれストリーム処理プロセス１１０−１〜１１０−５で稼働すると定義されるクエリを表す。【００４８】
例えば、分析クエリ５１３はＱ３、分析クエリ５１４はＱ１２とＱ４を表す。実行ノード指定アノテーション５２０の構成要素であるアノテーション５２１〜５２５は、それぞれ分析クエリ５１１〜５１５の実行プロセス（ストリーム処理プロセス１１０−１〜１１０−５）に対応する。【００４９】
図６は、従来例を示し、場の生産ラインを監視する処理を、ストリーム処理プロセス間で連携する一例を示すブロック図である。【００５０】
ストリーム処理サーバ１００−１、１００−２、１００−３上で、それぞれストリーム処理プロセス１１０−１、１１０−２、１１０−３が稼働し、それぞれのストリーム処理サーバ１００が有するセンサ２００−１、２００−２、２００−３のデータを分析する。各センサ２００は、データの取得時刻を表すタイムスタンプと、検出した製造物のセンサ値の組合せを、ストリームタプル６３１〜６３９として生成し、ストリーム処理プロセス１１０−１、１１０−２、１１０−３に入力する。【００５１】
ストリーム処理プロセス１１０−１、１１０−２、１１０−３は、センサデータを分析して、それぞれ分析結果タプル６４１、６４２、６４３を生成し、自身の下流に位置するストリーム処理プロセス１１０に送信する。【００５２】
分析結果タプル６４１、６４２、６４３は、センサデータのタイムスタンプと、分析結果の組合せである。このとき、各ストリーム処理プロセス１１０は、センサデータのみならず、自身の上流に位置するストリーム処理プロセス１１０の分析結果タプルと付き合わせて分析する。また、上流プロセスより受信した過去の分析結果タプルは、この付き合わせ対象として必要な分だけ確保しておく。分析結果タプル６４４〜６４７は、このために確保されたタプルである。【００５３】
センサ２００−１、２００−２、２００−３は、生産ライン６５０を流れる製造物６５１〜６５５のデータを取得する。センサ２００−１でデータを取得してから、センサ２００−２でデータを取得するまでの時間差は３秒であり、センサ２００−２でデータを取得してから、センサ２００−３でデータを取得するまでの時間差は２秒である。【００５４】
例えば、センサ２００−１が、製造物６５１のデータを取得した時刻は９：０５：０５で、その際にストリームタプル６３５を生成している。その後、センサ２００−２が、該製造物６５１のデータを取得した時刻は９：０５：０８で、その際にストリームタプル６３８を生成している。その後、センサ２００−３が、該製造物６５１のデータを取得した時刻は９：０５：１０で、その際にストリームタプル６３９を生成している。【００５５】
時刻９：０５：１０において、センサ２００−１、２００−２、２００−３が、それぞれ製造物６５５、６５３、６５１のデータを取得し、それぞれストリームタプル６３１、６３６、６３９を生成している。各ストリームタプルに付帯する四角が示す値は、実時間を表す。例えば、各センサ２００は、小数点以下第二位までの精度で正秒のタイミングでデータを取得する。【００５６】
各ストリームタプルが、ストリーム処理プロセス１１０で分析されてから、次のストリーム処理プロセスに分析結果が到達するまで、実時間で約０．３秒の時間が掛るとする。例えば、９：０５：１０．００にセンサ２００−１が生成したストリームタプル６３１が、ストリーム処理プロセス１１０−１で分析され、その結果である分析結果タプル６４１が、ストリーム処理プロセス１１０−２に到達した時刻は、９：０５：１０．３２となる。この例では、０．３２秒がストリーム処理サーバ１００−１の処理に要した時間となる。【００５７】
ここで、ストリーム処理プロセス１１０−２では、９：０５：１０．００にセンサ２００−２が生成したストリームタプル６３６の分析を、９：０５：１０．００から開始することはできない。なぜならば、ストリームタプル６３６のタイムスタンプ９：０５：１０以降のタイムスタンプを含む、ストリーム処理プロセス１１０−１からの分析結果タプルを、９：０５：１０．００の時点では受取っていないためである。【００５８】
ストリーム処理プロセス１１０−２がストリーム処理プロセス１１０−１より直近に受取った分析結果タプルは６４５であり、そのタイムスタンプは９：０５：０９である。そのため、ストリーム処理プロセス１１０−１からは、９：０５：０９〜９：０５：１０のタイムスタンプを含む分析結果タプルが到来する可能性があり、ストリームタプル６３６を、その分析結果タプルよりも先に処理することを禁止する必要がある。時刻順に従った、厳密に再現性のあるストリーム処理を実現するためには、このような実行制御が必要になる。【００５９】
ストリーム処理サーバ１００−２で、９：０５：１０の分析を開始するタイミングは、分析結果タプル６４１が到来する９：０５：１０．３２まで保留されるため、９：０５：１０の分析結果タプル６４２がストリーム処理プロセス１１０−２から１１０−３に到達するタイミングは、さらに約０．３秒あとの９：０５：１０．６１になる。同様にストリーム処理プロセス１１０−３による９：０５：１０の分析結果タプル６４３が、後段のストリーム処理プロセスに１１０−４到達するタイミングは、更に約０．３秒あとの９：０５：１０．９３になる。このように、生産ラインが長くなる程、分析結果が出力されるタイミングは遅れることになる。【００６０】
図７は、自動車の車両間通信を実現する、従来のストリーム処理プロセス間連携の一例を示すブロック図である。この従来例では、車載マイクロコンピュータであるストリーム処理サーバ１００−１、１００−２上で、それぞれストリーム処理プロセス１１０−１、１１０−２が本実施例と同様に稼働し、それぞれのストリーム処理サーバ１００が搭載されている車両の、各種センサで取得した位置情報や、速度の情報を表す位置・速度タプル７３１、７３２を分析して、数秒後の未来の自車の予測位置を表す予測位置タプル７７０、７７１を算出し、他車に転送する。このとき、各ストリーム処理プロセス１１０は、位置・速度タプル７３１、７３２のみならず、他車の予測位置タプル７７０、７７１と付き合わせて分析する。【００６１】
位置・速度タプル７３１、７３２は正秒毎に取得され、取得時のタイムスタンプが付与されてストリーム処理プロセス１１０に入力される。また、予測位置タプル７７０、７７１にも、予測の元となった位置・速度タプル７３１、７３２と同じ値のタイムスタンプが付与される。【００６２】
ここで、ストリーム処理プロセス１１０−１は、９：０５：１０．００に取得した、９：０５：１０のタイムスタンプを含む位置・速度タプル７３１と、ストリーム処理プロセス１１０−２から到来する、同じく９：０５：１０のタイムスタンプを含む予測位置タプル７７０とを突き合わせて、予測位置タプルを生成し、ストリーム処理プロセス１１０−２に送付する。【００６３】
従って、この処理は、予測位置タプル７７０がストリーム処理プロセス１１０−２から１１０−１に到来するまで保留されることになる。一方、ストリーム処理プロセス１１０−２においても、ストリーム処理プロセス１１０−１から予測位置タプル７７１が到来するまで、位置・速度タプル７３２の処理が保留されるため、予測位置タプル７７０の生成が保留される。【００６４】
このように、従来例のストリーム処理プロセス１１０−１とストリーム処理プロセス１１０−２は、９：０５：１０の分析処理を進めるにあたって、お互いに９：０５：１０の分析結果を必要とする、という相互依存関係を含んでいるため、ハングアップ（またはデッドロック）する、という問題があった。【００６５】
図８は、可変遅延演算を挿入した再帰クエリを工場の生産ラインの監視処理に適用した例を示すブロック図である。ストリーム処理分散実行システム８０１は、ストリーム処理サーバ１００−１〜１００−５を管理しており、各ストリーム処理サーバ１００では、それぞれストリーム処理プロセス１１０−１〜１１０−５が稼働する。【００６６】
ストリーム処理分散実行システム８０１が受け付けるクエリ定義１３２は、クエリ定義１３２を含むファイルで定義されるクエリグラフが、入力ストリーム８１１と、分析クエリ８１２と、および可変遅延演算８１３を含んでいる。そして、クエリ定義１３２は、分析クエリ８１２の出力が、可変遅延演算８１３を介して、再び分析クエリ８１２自身への入力となる、再帰クエリ（再帰ループ８２０）を構成している必要がある。【００６７】
ストリーム処理分散実行システム８０１は、例えば、ストリーム処理分散実行管理サーバ１００１がクエリ定義１３２を受取ると、クエリ分配部１００２が、ストリーム処理プロセス１１０−１〜１１０−５に分配する。この時点では、各ストリーム処理プロセス１１０上のクエリグラフはそれぞれ独立して稼働する。【００６８】
この状態において、入力側サーバ識別子（または、データソースの識別子）、出力側サーバ識別子（またはデータ受信機１４０の識別子）、遅延時間、およびタイムスタンプ（実行開始時刻）から構成される構成変更命令８１５が、例えば、ホスト計算機１３０よりストリーム処理分散実行システム８０１に投入されると、ストリーム処理サーバ１００ではクエリグラフの再設定が行われる。【００６９】
すなわち、ストリーム処理プロセス１１０が構成変更命令８１５を受信すると、クエリグラフの再帰ループ８２０は切断され、入力ストリーム８１１と、分析クエリ８１２と、可変遅延演算８１３が一方通行で並ぶ形となる。そして、可変遅延演算８１３の出力は、下流のクエリグラフの分析クエリ８１２へ入力される。【００７０】
入力側サーバ識別子で示されたストリーム処理サーバ１００上で稼働するストリーム処理プロセス１１０から、出力側サーバ識別子で示されたストリーム処理サーバ１００上で稼働するストリーム処理プロセス１１０へと、タプルの送信経路が設定される。【００７１】
入力側のストリーム処理プロセス１１０が生成した分析結果タプルを、出力側のストリーム処理プロセス１１０へ送信する際には、該タプルのタイムスタンプを、該遅延時間の値を加算したタイムスタンプに変更する。なお、可変遅延演算の効果は、図６、図７の従来例との対比について後述する。【００７２】
図５に示したクエリグラフと等価な分析を実行するクエリグラフを、クエリ定義１３２によって定義することができる。この定義は、各分析クエリ８１２が、入力ストリームと、他の分析クエリ８１２の結果タプルを処理して、分析結果を生成し、他の分析クエリ８１２（下流の分析クエリ）に出力することを表現している。分析クエリ８１２間の接続関係は、構成変更命令８１５によって動的に定義または変更することが可能である。【００７３】
以上のように、再帰クエリを含むクエリ定義１３２を各ストリーム処理プロセス１１０に配布しておき、構成変更命令８１５を与えることで分析クエリ８１２に与える入力ストリーム８１１と、可変遅延演算８１３の出力先を決定することが可能となる。【００７４】
これにより、ストリーム処理プロセス１１０を停止させることなく、分析クエリ８１２の入出力関係と遅延時間を動的に変更することが可能となり、ストリーム処理システムの可用性を向上させることができる。そして、構成変更命令８１５によって、多数のストリーム処理プロセス１１０の分析クエリ８１２の入出力関係と遅延時間を随時変更することが可能となるため、生産設備の変更や製品の変更などに容易に対処することが可能となる。【００７５】
さらに、各ストリーム処理プロセス１１０に配布するクエリ定義１３２は少なくとも一つあれば良いので、リアルタイム分析アプリケーションの開発を容易に行うことができる。【００７６】
なお、構成変更命令８１５の入力側サーバ識別子は、他のストリーム処理サーバ１００やセンサ２００やデータソース１２０等、１以上の計算機やデバイスを指定することができる。構成変更命令８１５の出力側サーバ識別子は、タプルを送信するストリーム処理サーバ１００やデータ受信機１４０など１以上の計算機を指定することができる。なお、データソース１２０は、計算機やセンサネットワークなどタプルを出力する装置で構成すればよい。【００７７】
図９Ａ、図９Ｂは、可変遅延演算を挿入した再帰クエリの一例を示すスクリプトである。図９Ａのクエリ９００の冒頭は、センサ２００−１〜２００−４が取得するデータを入力ストリームとして定義する、ＳＸのｒｅｇｉｓｔｅｒ  ｓｔｒｅａｍ句が記載されている。【００７８】
クエリＱＸは、ストリーム処理プロセス１１０−１〜１１０−５で稼働する分析クエリを定義する。クエリＱＸＤは、クエリＱＸの結果に対して可変遅延演算を付加する処理を定義しており、＜ｖａｒｉａｂｌｅ＞という記述が、固定された遅延時間の定義を持たない、可変遅延演算の操作を定義している。【００７９】
クエリＱＸは、クエリＱＸＤの結果、即ちクエリＱＸ自身の結果を遅延させたストリームと、ストリームＳＸの両者を入力として、分析処理を生成し、クエリＱＸＤを介して再びクエリＱＸ自身の入力としている、という再帰的な処理を定義している。クエリ４００と等価な分析を実行するクエリを、クエリ９００によって定義することができる。【００８０】
図９Ｂのクエリ９０１は、分析クエリが多数のｒｅｇｉｓｔｅｒ  ｑｕｅｒｙ句で構成される例を示している。クエリＱＡからクエリＱＸまでの一連のクエリによって分析クエリが構成され、最後のクエリＱＸの結果が、クエリＱＸＤで遅延されて、再びクエリＱＡの結果として入力されている。このようなクエリも、クエリ定義１３２として、ストリーム処理分散実行システム８０１に投入することが可能である。【００８１】
図１０は、ストリーム処理プロセス１１０の構成を示すブロック図である。ストリーム処理分散実行システム８０１は、ストリーム処理分散実行管理サーバ１００１と、ストリーム処理プロセスを実行する複数のストリーム処理サーバ１００を有する。ストリーム処理分散実行管理サーバ１００１は、クエリ登録インタフェース１３１などを介して、例えば、ホスト計算機１３０からクエリ定義１３２を受取り、ストリーム処理プロセス１１０に分配するクエリ分配部１００２、および、例えば、ホスト計算機１３０から構成変更命令８１５を受信し、各ストリーム処理プロセス１１０に分配する構成変更命令分配部１００３と、ストリーム処理サーバ１００の識別子とアドレスを予め設定したサーバテーブル１００４とを含む。【００８２】
構成変更命令８１５の入力側サーバ識別子、および出力側サーバ識別子において、それぞれストリーム処理サーバ１００−１、およびストリーム処理サーバ１００−２が指定されている場合を想定する。【００８３】
ストリーム処理プロセス１１０−１、１１０−２は、それぞれ、クエリ処理エンジン（クエリ処理部）１０１１、１０３１と、クエリ解析部１０１２、１０３２と、タプルの入力アダプタ１０１６、１０３６と、および構成変更命令受信部１０２２、１０４１とを有する。【００８４】
クエリ処理エンジン１０１１、１０３１は、それぞれ、クエリグラフ１０１３、１０３３と、クエリグラフの結果タプルを保存する結果タプルキュー１０１４、１０３４、およびクエリグラフの実行時間を管理する、実行時刻管理部１０１５、１０３５を有する。【００８５】
クエリ解析部１０１２、１０３２は、クエリ分配部１００２からクエリ定義１３２に記載されたクエリ定義を受取り、該クエリ定義をパースしてクエリグラフを構築し、それぞれ、クエリ処理エンジン１０１１、１０３１の中に、クエリグラフ１０１３、１０３３として投入する。【００８６】
構成変更命令受信部１０２２、１０４１は、入力側のストリーム処理プロセス１１０−１に位置する構成変更命令受信部１０２２と、出力側のストリーム処理プロセス１１０−２に位置する構成変更命令受信部１０４１とで処理内容が異なる。【００８７】
出力側のストリーム処理プロセス１１０−２の構成変更命令受信部１０４１は、入力側サーバ識別子にて指定されたサーバ１００−１が出力するタプルを受け付けるように、入力アダプタ１０３６と出力アダプタ１０１７との間の通信接続を確立する。一方、入力側のストリーム処理プロセス１１０−１の構成変更命令受信部１０２２も、出力側サーバ識別子にて指定されたサーバ１００−２に対してタプルを出力するように、入力アダプタ１０３６と出力アダプタ１０１７との間の通信接続を確立する。【００８８】
また、入力側のストリーム処理プロセス１１０−１は、構成変更命令受信部１０２２が受信した構成変更命令８１５をタイムスタンプ順に保存する、構成変更命令キュー１０２３と、結果タプルキュー１０１４に保存されている結果タプルのタイムスタンプをシフトさせて、出力アダプタ１０１７に他サーバへの送信を依頼する、可変遅延演算実行部１０２４、および出力先別に異なる遅延時間幅を保存する、出力先別遅延時間管理表１０２５とを有する。なお、可変遅延演算実行部１０２４は、図８のクエリ定義１３２に示した可変遅延演算８１３に対応する。【００８９】
なお、出力アダプタ１０１７と入力アダプタ１０１６、１０３６の設定は、サーバテーブル１００４を参照してサーバ識別子に対応するアドレスを取得すれば良い。サーバテーブル１００４は、サーバ識別子とＩＰアドレスを一つのエントリとして構成すれば良い。また、ストリーム処理分散実行管理サーバ１００１は、各ストリーム処理サーバ１００にサーバテーブル１００４を配布して、各ストリーム処理プロセス１１０がサーバテーブル１００４を参照するようにしてもよい。【００９０】
さらに、ストリーム処理プロセス１１０は、例えば、ホスト計算機１３０から投入される構成変更命令８１５を受信するのではなく、分析対象の変化などに追従して自律的に構成変更を決定する、構成変更自律決定部１０２１を有する。【００９１】
構成変更自律決定部１０２１は、決定した構成変更を構成変更命令８１５として生成し、入力側と出力側の双方に備わる構成変更命令受信部１０２２、１０４１に向けて送信する。構成変更自律決定部１０２１は、入力側と出力側のどちらのストリーム処理プロセス１１０に配置されても構わない。【００９２】
構成変更自律決定部１０２１が構成変更を決定するために用いるアルゴリズムは、特定の手順には固定されておらず、分析対象に合せて交換可能とする。そのため、構成変更自律決定部１０２１は、構成変更決定アルゴリズム向けのＡＰＩを有しており、該ＡＰＩに従って実装された所定のアルゴリズムを組込み可能とする。【００９３】
以上の構成により、ストリーム処理サーバ１００−１は、受け付けた入力ストリームを分析クエリで処理し、さらに、遅延演算を行ったタプルを結果タプル１４１としてストリーム処理サーバ１００−２へ出力する。ストリーム処理サーバ１００−２は、入力ストリームに加えてストリーム処理サーバ１００−１からの結果タプル１４１を受け付けて分析クエリで処理を行い、結果タプル１４１を出力する。【００９４】
なお、図示の例では、ストリーム処理サーバ１００−２の遅延演算実行部と出力アダプタ等は機能しないため図示されていない。しかし、ストリーム処理サーバ１００−２の下流で、さらに分析クエリを実行する場合には、可変遅延演算実行部と出力アダプタ等が有効化される。【００９５】
また、ストリーム処理プロセス１１０のクエリ解析部１０１２と、構成変更自律決定部１０２１と、構成変更命令受信部１０２２と、可変遅延演算実行部１０２４と、クエリ処理エンジン１０２６の各機能部はプログラムとしてメモリ１０３にロードされる。【００９６】
ＣＰＵ１０１は、各機能部のプログラムに従って処理することによって、所定の機能を提供する機能部として稼働する。例えば、ＣＰＵ１０１は、クエリ解析プログラムに従って処理することでデバイス選択クエリ解析部１０１２として機能する。他のプログラムについても同様である。さらに、ＣＰＵ１０１は、各プログラムが実行する複数の処理のそれぞれの機能を提供する機能部としても稼働する。計算機及び計算機システムは、これらの機能部を含む装置及びシステムである。【００９７】
図１１は、構成変更命令８１５の一例を示すブロック図である。構成変更命令８１５には、当該命令が９：０５：１０以降のタイムスタンプを含む結果タプルに対して適用開始となること、入力側サーバ、出力側サーバがそれぞれ１００−１、１００−２であること、および遅延時間が３秒であることが示されている。【００９８】
ストリーム処理分散実行管理サーバ１００１の構成変更命令分配部１００３は、構成変更命令８１５を、入力側のストリーム処理サーバ１００−１の構成変更命令受信部１０２２に配送する入力側構成変更命令１１０１と、出力側のストリーム処理サーバ１００−２の構成変更命令受信部１０４１に配送する出力側構成変更命令１１０２に分解する。【００９９】
ストリーム処理サーバ１００−１、１００−２では、構成変更命令８１５を受け付けると、指定された時刻に入出力の関係と遅延時間を動的に切り替えることが可能となる。【０１００】
図１２Ａ〜図１２Ｄは、可変遅延演算実行部で処理するデータの一例を示す図である。図１２Ａは、結果タプルキュー１０１４の一例を示す図である。結果タプルキュー１０１４には、結果タプル１２１１、１２１２、１２１３、１２１４が格納されている状態を示す。【０１０１】
図１２Ｂは、構成変更命令キュー１０２３の一例を示す図である。構成変更命令キュー１０２３には、入力側構成変更命令１１０１、１２３１が格納された例を示す。図示の例では、９：０５：１０で出力側がストリーム処理サーバ１００−２へ切り替えられて遅延時間が３秒に設定され、９：０５：２０で出力側がストリーム処理サーバ１００−３へ切り替えられて遅延時間が４秒に変更される例を示している。【０１０２】
図１２Ｃは、出力先別遅延時間管理表１０２５の一例を示す図である。出力先別遅延時間管理表１０２５には、二つのエントリが存在し、出力先１２０１、および遅延時間１２０２の値が、それぞれ図の状態となっている。【０１０３】
図１２Ｄは、可変遅延演算実行部１０２４の状態を示す図である。可変遅延演算実行部１０２４は、結果タプルキュー１０１４の先頭タプル１２１１を取出して、出力先別遅延時間管理表１０２５の遅延時間１２０２のカラムの値に合せて、タプル１２１１のタイムスタンプに、それぞれ３秒と５秒を加えた２つのタプル１２２１、１２２２を生成し、出力先１２０１のカラムが示すストリーム処理サーバ１００−３、１００−４（図中サーバ３、サーバ４）に、それぞれ送付する。【０１０４】
ここで、可変遅延演算実行部１０２４は、構成変更命令キュー１０２３の先頭に位置する、構成変更命令１１０１のタイムスタンプ９：０５：１０と、結果タプル１２１１のタイムスタンプ９：０５：０９を比較し、前者の方が未来であったため、構成変更命令１１０１はキューに保留される。この後、可変遅延演算実行部１０２４は、結果タプル１２１２を処理する際には、当該タプルのタイムスタンプ９：０５：１０が、構成変更命令１１０１のタイムスタンプ９：０５：１０以降であるため、構成変更命令１１０１の内容を、出力先別遅延時間管理表１０２５に格納するため、出力先としてサーバ２（ストリーム処理サーバ１００−２）のエントリを加え、３エントリとする。その結果として、タプル１２１２を３つのストリーム処理サーバ１００−２〜１００−４に送付する。【０１０５】
図１３は、可変遅延演算実行部１０２４で行われる処理の一例を示すフローチャートである。図１３の処理は、ストリーム処理サーバ１００−１の可変遅延演算実行部１０２４における、一つのタプルの処理の流れを示しており、ステップ１３００から開始して、ステップ１３１１で終了する。この処理は、タプルが到着する度に実行される。【０１０６】
まず、ステップ１３０１にて、可変遅延演算実行部１０２４は、結果タプルキュー１０１４から先頭タプルを取得する。続くステップ１３０２にて、可変遅延演算実行部１０２４は、構成変更命令キュー１０２３に構成変更命令８１５が存在するか否かを判定する。
構成変更命令８１５が存在しない場合にはステップ１３０７に進む。【０１０７】
一方、構成変更命令８１５が存在する場合には、可変遅延演算実行部１０２４がステップ１３０３にて、上記ステップ１３０１で結果タプルキュー１０１４から取得したタプルのタイムスタンプと、構成変更命令キュー１０２３の先頭に位置する構成変更命令のタイムスタンプを比較する。前者の方が過去である場合は、ステップ１３０７に進む。【０１０８】
それ以外の場合は、ステップ１３０４にて、可変遅延演算実行部１０２４は、当該構成変更命令の出力側のサーバ識別子にマッチするエントリが、出力先別遅延時間管理表１０２５に存在するか否かを判定し、存在する場合には、ステップ１３０５に進む。一方、出力側のサーバ識別子にマッチするエントリが、出力先別遅延時間管理表１０２５に存在しない場合には、ステップ１３０６へ進む。【０１０９】
ステップ１３０５では、可変遅延演算実行部１０２４が、当該エントリを、当該構成変更命令の内容に更新する。一方、ステップ１３０６の、存在しない場合には、可変遅延演算実行部１０２４が当該構成変更命令の内容を、出力先別遅延時間管理表１０２５に、新規エントリとして登録する。そして、ステップ１３０５、１３０６の処理が完了すると、ステップ１３０２に戻って上記処理を繰り返す。【０１１０】
ステップ１３０７からステップ１３１０の処理は、可変遅延演算実行部１０２４が出力先別遅延時間管理表１０２５に存在するエントリの数だけ繰り返す。まず、ステップ１３０８にて、可変遅延演算実行部１０２４が上記ステップ１３０１で結果タプルキュー１０１４から取得したタプルのコピーを生成し、当該エントリの遅延時間１２０２の分だけ、タイムスタンプをシフトさせる。【０１１１】
そして、ステップ処理１３０９にて、可変遅延演算実行部１０２４は当該エントリの出力先１２０１に、前記でコピーしたタプルを送信する。ステップ１３０７からステップ１３１０のループを抜けると、可変遅延演算実行部１０２４はステップ１３１１に到達して処理を終了する。【０１１２】
上記処理によって、結果タプルキュー１０１４の結果タプル１４１のタイムスタンプが、構成変更命令キュー１０２３の構成変更命令のタイムスタンプであれば、可変遅延演算実行部１０２４は構成変更命令の出力先を加えて、結果タプル１４１を送信する。これにより、ストリーム処理プロセス１１０を停止することなく、入出力の関係を動的に変更することが可能となるのである。【０１１３】
図１４は、工場の生産ラインを監視する、ストリーム処理プロセス１１０間の、遅延演算を介した連携の一例を示すブロック図である。前記従来例の図６に対して、ストリーム処理プロセス１１０−１、１１０−２に、それぞれ遅延演算１４０１、１４０２が挿入されている。【０１１４】
それぞれの遅延演算に設定されている遅延時間３秒、および２秒は、センサ２００−１でデータを取得してからセンサ２００−２でデータを取得するまでの時間差、および、センサ２００−２でデータを取得してからセンサ２００−３でデータ取得するまでの時間差に合せて決められている。【０１１５】
９：０５：０７にセンサ２００−１が生成したタプル６３３は、ストリーム処理プロセス１１０−１で分析され、結果タプル６４４として出力される。このとき、分析結果タプル６４４のタイムスタンプは、遅延演算によって、９：０５：０７から３秒後の９：０５：１０と定まる。同タプル６４４は、実時間において９：０５：０７．００から約０．３秒経過した９：０５：０７．３３に、ストリーム処理プロセス１１０−２に到達する。【０１１６】
その後、９：０５：１０．００に、センサ２００−２がストリームタプル６３６を生成した際には、９：０５：１０のタイムスタンプを含む、分析結果タプル６４４と、ストリームタプル６３６の両方が揃っているため、ストリーム処理プロセス１１０−２は、９：０５：１０．００の即時に分析を開始することが可能である。従って、この分析結果タプル６４２は、その約０．３秒後の９：０５：１０．３１に、ストリーム処理プロセス１１０−３に到達することができる。【０１１７】
また、遅延演算１４０１の遅延時間を、ストリーム処理プロセス１１０間のデータ取得の時間差である３秒に合せたことで、同一の製造物ｃ（６５３）について、ストリーム処理プロセス１１０−１での分析結果とセンサデータが揃うため、ストリーム処理プロセス１１０−２では、９：０５：１０の分析結果タプル６４２を出力した後は、タプル６４４を保持しておく必要はなくなる。【０１１８】
ストリーム処理プロセス１１０−２の遅延演算１４０２によって、ストリーム処理プロセス１１０−３での分析処理においても同様の効果があるため、分析結果タプル６４３も、ストリームタプル６３９が生成された９：０５：１０．００の約０．３秒後である９：０５：１０．３０に、ストリーム処理プロセス１１０−３の後段のプロセスに到達することができる。【０１１９】
このように、本実施例によれば、生産ラインが長くなって同一の物体を検出する時間差が増大した場合でも、分析結果の出力タイミングの遅れが増加することは回避できる。そして、遅延時間を付加した場合でもタプルの時刻順を維持することが可能となって、ストリーム処理の一貫性の保証と、レイテンシの削減を両立させることができるのである。【０１２０】
図１５、図１６は、自動車の車両間通信を実現する、ストリーム処理プロセス１１０間の、遅延演算を介した連携の一例を示すブロック図である。前記従来例の図７に対して、図１５、図１６では、ストリーム処理プロセス１１０−１、１１０−２に、それぞれ遅延演算１５０１、１５０２が挿入されている。図１５で設定されている遅延時間２秒は、図１６では車両間の距離が縮まるため１秒に更新されている。【０１２１】
ストリーム処理プロセス１１０−１には、自車の位置・速度タプル７２０、７２１、７２２が、それぞれ９：０５：０６．００、９：０５：０７．００、９：０５：０８．００のタイミングで入力される。また、ストリーム処理プロセス１１０−２からは、他車の時刻ｔ０、ｔ０、ｔ１における予測位置を示す、予測位置タプル７４３、７４４、７４５が、それぞれ９：０５：０６．２１、９：０５：０７．２４、９：０８．２２のタイミングで到来する（ストリーム処理プロセス１１０−２において位置・速度タプルが生成されてから分析し、ストリーム処理プロセス１１０−１に到達するまで約０．２秒かかる）。また、遅延演算１５０２によって、それぞれのタイムスタンプは、９：０５：０８、９：０５：０９、９：０５：１０に変更されている。【０１２２】
ここで、ストリーム処理プロセス１１０−１にて、タイムスタンプ９：０５：０８の位置・速度タプル７２２が生成された９：０５：０８．００のタイミングには、同じタイムスタンプ９：０５：０８を含む、ストリーム処理プロセス１１０−２からの予測位置タプル７４３が既に到達している。したがってストリーム処理プロセス１１０−１では、タイムスタンプ９：０５：０８の分析処理として両タプルから時刻ｔ１の予測位置を算出し、遅延演算１５０１によりタイムスタンプを９：０５：１０にシフトさせて、予測位置タプル７４２として、ストリーム処理プロセス１１０−２に対して送付することが可能となる。ここで予測位置を割り出す時刻ｔ１が、遅延後のタイムスタンプ９：０５：１０を起点として、より未来であるほど、予測値としての価値が高いことになる。【０１２３】
ストリーム処理プロセス１１０−２側での分析においても同様の効果があるため、図７に示した従来例のようにハングアップを起こさずに、処理を進めることが可能となる。車両３００同士が接近した場合は、予測の効果を損なわないように、予測時刻ｔｘと、予測位置タプルに付与される遅延後のタイムスタンプの時間差を保つために、遅延時間を短縮する必要がある。【０１２４】
図１６の例では、遅延時間を２秒から１秒に短縮し、タイムスタンプ９：０５：０９の位置・速度タプル７２６から、タイムスタンプ９：０５：１０の予測位置タプル７４６を生成している。即ち、先に出力した予測位置タプル７４２と、タイムスタンプが重複することになるが、分析クエリの中で予測値の修正として扱うことができるので、この重複は特に問題にはならない。【０１２５】
以上のように、本実施例１によれば、ストリーム処理の再現性を維持しながら、クエリ定義を容易に行うことが可能となって、処理の停止を伴わない動的なクエリ間の接続関係の変更を実現できる。さらに、位置に応じて取得されたセンサデータが利用される時刻を見込んで遅延時間を設定することにより、レイテンシ増加の抑制および相互依存関係を含むクエリを実現することができる。
【実施例２】
【０１２６】
図１７は、本発明の実施例２を示し、ストリーム処理分散実行システム８０１のブロック図である。実施例２では、前記実施例１のストリーム処理分散実行管理サーバ１００１の機能をストリーム処理サーバ１００−１に組み込んだもので、その他の構成は前記実施例１と同様である。【０１２７】
ストリーム処理サーバ１００−１では、ストリーム処理プロセス１１０−１に加えて、クエリ分配部１００２と、構成変更命令分配部１００３と、サーバテーブル１００４が稼働する。【０１２８】
クエリ分配部１００２は、前記実施例１と同様に、ホスト計算機１３０からクエリ定義１３２を受け付けると内容に応じて他のストリーム処理プロセス１１０にクエリを分配する。また、構成変更命令分配部１００３も前記実施例１と同様に、ホスト計算機１３０から構成変更命令８１５を受け付けると、サーバテーブル１００４を参照して指定されたストリーム処理サーバ１００−１に構成変更命令８１５を分配する。【０１２９】
＜まとめ＞
以上のように、本実施例１、２では、分散処理において、ストリーム処理の再現性を維持しながら、クエリ定義の容易化と、処理停止を伴わない動的なクエリ間の接続関係の変更と、レイテンシ増加の抑制、および相互依存関係を含むクエリの実現を可能とする。【０１３０】
従って、ＩｏＴのセンサデータ分析に基づく機器の監視や制御、移動体間の協調処理などのように、エッジ（ストリーム処理サーバ１００）間での分析処理の分散実行にも適用可能な、ストリーム処理の一貫性の保証と、低レイテンシ処理を両立する、リアルタイム分析アプリケーションの開発を容易に行うことができる。また、分散処理の最適な構成を実行時に変更可能とするように、クエリの分散実行における位置の透過性を保証し、アプリケーションの開発とＳＩ（Ｓｙｓｔｅｍ  Ｉｎｔｅｇｒａｔｉｏｎ）作業の工程を分離し、アプリケーションの可搬性（または汎用性）を向上させることができる。【０１３１】
以上、本発明の実施例について説明したが、これらは本発明を説明するための例示であり、本発明の適応範囲を例示した形態のみに限定するものではない。また、前述した実施形態のいかなる組み合わせも本発明の実施形態となり得る。【０１３２】
なお、本発明は上記した実施例に限定されるものではなく、様々な変形例が含まれる。例えば、上記した実施例は本発明を分かりやすく説明するために詳細に記載したものであり、必ずしも説明した全ての構成を備えるものに限定されるものではない。また、ある実施例の構成の一部を他の実施例の構成に置き換えることが可能であり、また、ある実施例の構成に他の実施例の構成を加えることも可能である。また、各実施例の構成の一部について、他の構成の追加、削除、又は置換のいずれもが、単独で、又は組み合わせても適用可能である。【０１３３】
また、上記の各構成、機能、処理部、及び処理手段等は、それらの一部又は全部を、例えば集積回路で設計する等によりハードウェアで実現してもよい。また、上記の各構成、及び機能等は、プロセッサがそれぞれの機能を実現するプログラムを解釈し、実行することによりソフトウェアで実現してもよい。各機能を実現するプログラム、テーブル、ファイル等の情報は、メモリや、ハードディスク、ＳＳＤ（Ｓｏｌｉｄ  Ｓｔａｔｅ  Ｄｒｉｖｅ）等の記録装置、または、ＩＣカード、ＳＤカード、ＤＶＤ等の記録媒体に置くことができる。【０１３４】
また、制御線や情報線は説明上必要と考えられるものを示しており、製品上必ずしも全ての制御線や情報線を示しているとは限らない。実際には殆ど全ての構成が相互に接続されていると考えてもよい。
【図１】
0006674039.tif 000002
【図２】
0006674039.tif 000003
【図３】
0006674039.tif 000004
【図４】
0006674039.tif 000005
【図５】
0006674039.tif 000006
【図６】
0006674039.tif 000007
【図７】
0006674039.tif 000008
【図８】
0006674039.tif 000009
【図９Ａ】
0006674039.tif 000010
【図９Ｂ】
0006674039.tif 000011
【図１０】
0006674039.tif 000012
【図１１】
0006674039.tif 000013
【図１２Ａ】
0006674039.tif 000014
【図１２Ｂ】
0006674039.tif 000015
【図１２Ｃ】
0006674039.tif 000016
【図１２Ｄ】
0006674039.tif 000017
【図１３】
0006674039.tif 000018
【図１４】
0006674039.tif 000019
【図１５】
0006674039.tif 000020
【図１６】
0006674039.tif 000021
【図１７】
0006674039.tif 000022