(57)【要約】  時刻が含まれた入力データで構成されるストリームデータを処理するストリームデータ処理装置を提供するもので、前記ストリームデータを構成する入力データを受信するデータ入力部と、入力データを時刻順に処理する前記入力データの項目をデータセットとして指定する第１のキーと、前記ストリームデータ定義及びクエリ定義を受け付けて前記入力データを処理するオペレータを生成するクエリ登録部と、前記データセット毎に前記入力データを処理するオペレータを決定し、当該オペレータで前記入力データを処理した結果を出力するデータ実行部と、を有し、前記データ入力部が、前記第１のキーで指定された項目で、前記受信した入力データを前記データセット毎に分類し、前記データセット毎に前記入力データを時刻順にソートして入力ストリームを生成し、データ実行部は、前記入力ストリームをデータセット毎に前記オペレータで処理する。
2012046316.tif 000002
【特許請求の範囲】
【請求項１】
プロセッサとメモリを備えたストリームデータ処理装置で、時刻が含まれた入力データで構成されるストリームデータを受信し、予め登録されたクエリに従って前記ストリームデータの処理を実行するストリームデータ処理方法であって、
前記ストリームデータ処理装置は、
前記ストリームデータを構成する複数の入力データを受信するデータ入力部と、
前記入力データを時刻順に処理する前記入力データの項目をデータセットとして指定する第１のキーと、前記ストリームデータの定義と、前記クエリの定義を受け付けて前記入力データを処理するオペレータを生成するクエリ登録部と、
前記データセット毎に前記入力データを処理するオペレータを決定し、当該オペレータで前記入力データを処理した結果を出力するデータ実行部と、を有し、
前記ストリームデータ処理方法は、
前記クエリ登録部が、前記第１のキーとクエリの定義及びストリームデータの定義を受け付けて、前記入力データに含まれる項目のうち、時刻順に処理するデータセットを設定する第１のステップと、
前記データ入力部が、前記入力データを受信して、入力ストリームを生成する第２のステップと、
前記データ実行部が、前記入力ストリームを受け付けて、当該入力ストリームに含まれる入力データを、前記データセット毎に前記オペレータで処理する第３のステップと、
前記オペレータが処理した結果を単一の出力ストリームとして生成する第４のステップと、を含むことを特徴とするストリームデータ処理方法。
【請求項２】
請求項１に記載のストリームデータ処理方法であって、
前記第３のステップは、
前記データ実行部が、前記第１のキーで指定された項目を用いて前記データセット毎に入力データを処理する実行領域を前記メモリ上に生成する第５のステップと、
前記データ実行部が、前記入力ストリームに含まれる入力データが属するデータセットを決定する第６のステップと、
前記データ実行部が、前記決定されたデータセットに対応する前記実行領域へ前記入力データを格納する第７のステップと、
前記データ実行部が、前記実行領域毎に前記オペレータで前記入力データを処理する第８のステップと、を含むことを特徴とするストリームデータ処理方法。
【請求項３】
請求項２に記載のストリームデータ処理方法であって、
前記第５のステップは、
前記データ実行部が、前記データセットに対応する入力データを初めて受け付けたときに前記実行領域を前記メモリ上に生成することを特徴とするストリームデータ処理方法。
【請求項４】
請求項２に記載のストリームデータ処理方法であって、
前記第８のステップは、
前記データ実行部が、前記入力データを処理する前記オペレータを前記実行領域に格納し、前記オペレータを実行する時刻を発火時刻として格納し、前記時刻が前記発火時刻となった入力データを実行し、前記オペレータの処理が終わった入力データと同一のデータセットかつ同一時刻に実行可能な次の入力データがある限り、当該オペレータで前記次の入力データを処理することを特徴とするストリームデータ処理方法。
【請求項５】
請求項２に記載のストリームデータ処理方法であって、
前記第８のステップは、
前記データ実行部が、前記入力データとオペレータ毎に実行可能か否かを示す情報を格納する実行可能データ情報を設定する第９のステップと、
前記データ実行部が、前記実行可能データ情報を参照して実行可能な入力データとオペレータを決定する第１０のステップと、
前記データ実行部が、前記実行した入力データとオペレータに対応する実行可能データ情報を更新する第１１のステップと、
を含むことを特徴とするストリームデータ処理方法。
【請求項６】
請求項１に記載のストリームデータ処理方法であって、
前記第１のステップは、
前記クエリ登録部が、前記データセットとして指定する第１のキーに加えて、前記入力データを時刻順にソートする前記入力データの項目を指定する第２のキーを受け付けて、
前記第２のステップは、
前記データ入力部が、前記第１のキーで指定された項目で、前記入力データを前記データセット毎に分類し、前記データセット毎に前記第２のキーに対応する入力データの項目を時刻順にソートして入力ストリームを生成する際に、前記第１のキーの値が同一の入力データのうち、前記第２のキーの値が異なる入力データを時刻順にソートすることを特徴とするストリームデータ処理方法。
【請求項７】
請求項１に記載のストリームデータ処理方法であって、
前記第１のステップは、
前記クエリ登録部が、前記第１のキーとクエリの定義及びストリームデータの定義を受け付けて、前記ストリームデータの定義を複製して複数の入力ストリームを定義し、前記入力データに含まれる項目のうち、時刻順に処理するデータセットと複数の入力ストリームの関係を設定し、
前記第２のステップは、
前記データ入力部が、前記入力データを受信して、前記第１のキーで指定された項目で、前記入力データを前記データセット毎に分類し、前記データセット毎に前記入力データを時刻順にソートして前記データセットと複数の入力ストリームの関係から複数の入力ストリームを生成し、
前記第３のステップは、
前記データ実行部が、前記複数の入力ストリームをそれぞれ受け付けて、当該入力ストリームに含まれる入力データを、前記入力ストリーム毎に前記オペレータで処理し、
前記第４のステップは、
前記複数の入力ストリームをオペレータが処理した結果を単一のキューに出力して出力ストリームを生成することを特徴とするストリームデータ処理方法。
【請求項８】
プロセッサとメモリを備えて、時刻が含まれた入力データで構成されるストリームデータを受信し、予め登録されたクエリに従って前記ストリームデータの処理を実行するストリームデータ処理装置であって、
前記ストリームデータを構成する複数の入力データを受信するデータ入力部と、
前記入力データを時刻順に処理する前記入力データの項目をデータセットとして指定する第１のキーと、前記ストリームデータの定義と、前記クエリの定義を受け付けて前記入力データを処理するオペレータを生成するクエリ登録部と、
前記データセット毎に前記入力データを処理するオペレータを決定し、当該オペレータで前記入力データを処理した結果を出力するデータ実行部と、を有し、
前記クエリ登録部は、
前記第１のキーとクエリの定義及びストリームデータの定義を受け付けて、前記入力データに含まれる項目のうち、時刻順に処理するデータセットを設定し、
前記データ入力部は、
前記入力データを受信して、入力ストリームを生成し、
前記データ実行部は、
前記入力ストリームを受け付けて、当該入力ストリームに含まれる入力データを、前記データセット毎に前記オペレータで処し、前記オペレータが処理した結果を単一の出力ストリームとして生成することを特徴とするストリームデータ処理装置。
【請求項９】
請求項８に記載のストリームデータ処理装置であって、
前記データ実行部は、前記第１のキーで指定された項目を用いて前記データセット毎に入力データを処理する実行領域を前記メモリ上に生成し、前記入力ストリームに含まれる入力データが属するデータセットを決定し、前記決定されたデータセットに対応する前記実行領域へ前記入力データを格納し、前記実行領域毎に前記オペレータで前記入力データを処理することを特徴とするストリームデータ処理装置。
【請求項１０】
請求項９に記載のストリームデータ処理装置であって、
前記データ実行部は、前記データセットに対応する入力データを初めて受け付けたときに前記実行領域を前記メモリ上に生成することを特徴とするストリームデータ処理装置。
【請求項１１】
請求項９に記載のストリームデータ処理装置であって、
前記データ実行部は、前記入力データを処理する前記オペレータを前記実行領域に格納し、前記オペレータを実行する時刻を発火時刻として格納し、前記時刻が前記発火時刻となった入力データを実行し、前記オペレータの処理が終わった入力データと同一のデータセットかつ同一時刻に実行可能な次の入力データがある限り、当該オペレータで前記次の入力データを処理することを特徴とするストリームデータ処理装置。
【請求項１２】
請求項９に記載のストリームデータ処理装置であって、
前記データ実行部は、前記入力データとオペレータ毎に実行可能か否かを示す情報を格納する実行可能データ情報を設定し、前記実行可能データ情報を参照して実行可能な入力データとオペレータを決定し、前記実行した入力データとオペレータに対応する実行可能データ情報を更新することを特徴とするストリームデータ処理装置。
【請求項１３】
請求項８に記載のストリームデータ処理装置であって、
前記クエリ登録部は、前記データセットとして指定する第１のキーに加えて、前記入力データを時刻順にソートする前記入力データの項目を指定する第２のキーを受け付け、
前記データ入力部は、前記第１のキーで指定された項目で、前記入力データを前記データセット毎に分類し、前記データセット毎に前記第２のキーに対応する入力データの項目を時刻順にソートして入力ストリームを生成する際に、前記第１のキーの値が同一の入力データのうち、前記第２のキーの値が異なる入力データを時刻順にソートすることを特徴とするストリームデータ処理装置。
【請求項１４】
請求項８に記載のストリームデータ処理装置であって、
前記クエリ登録部は、前記第１のキーとクエリの定義及びストリームデータの定義を受け付けて、前記ストリームデータの定義を複製して複数の入力ストリームを定義し、前記入力データに含まれる項目のうち、時刻順に処理するデータセットと複数の入力ストリームの関係を設定し、
前記データ入力部は、前記入力データを受信して、前記第１のキーで指定された項目で、前記入力データを前記データセット毎に分類し、前記データセット毎に前記入力データを時刻順にソートして前記データセットと複数の入力ストリームの関係から複数の入力ストリームを生成し、
前記データ実行部は、前記複数の入力ストリームをそれぞれ受け付けて、当該入力ストリームに含まれる入力データを、前記入力ストリーム毎に前記オペレータで処理し、前記複数の入力ストリームをオペレータが処理した結果を単一のキューに出力して出力ストリームを生成することを特徴とするストリームデータ処理装置。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、ストリームデータの処理方法及び装置に関する。
【背景技術】
【０００２】
時々刻々と到着する大量のデータをリアルタイム処理するデータ処理システムに対する要求が高まっている。例えば、株式自動売買、カープローブ、Ｗｅｂのアクセス監視、製造監視などを挙げることができる。【０００３】
従来、企業情報システムのデータ管理の中心にはデータベース管理システム（以下、ＤＢＭＳとする）が位置づけられていた。ＤＢＭＳは、処理対象のデータをストレージに格納し、格納したデータに対してトランザクション処理に代表される高信頼な処理を実現している。しかし、ＤＢＭＳでは、新たなデータが到着する度に、全てのデータに対して検索処理を施すため上記のリアルタイム処理の要求を満たすことは難しい。例えば、株取引を支援する金融アプリケーションを考えた場合、株価の変動にいかに迅速に反応できるかがシステムの最重要課題の一つである。しかし、上記従来のＤＢＭＳでは、データの検索処理が株価変動のスピードに追いつくことができず、ビジネスチャンスを逃してしまうことになりかねない。【０００４】
このようなリアルタイムデータ処理に好適なデータ処理システムとして、ストリームデータ処理システムが提案されている。例えば非特許文献１にストリームデータ処理システム"ＳＴＲＥＡＭ"が開示されている。ストリームデータ処理システムでは従来のＤＢＭＳとは異なり、まずクエリ（問合せ）をシステムに登録しておき、データの到来と共に該クエリが継続的に実行される。実行されるクエリがあらかじめ把握できるため、新たなデータが到着したら、それまでの処理結果からの差分のみを処理することで高速な処理が可能である。したがって、ストリームデータ処理によって、株取引などにおける高レートで発生するデータをリアルタイムに解析し、ビジネスに有効なイベントの発生を監視して活用することが可能になる。【０００５】
ストリームデータ処理では時刻順の入力データを前提とすることで、データ入力と同時に逐次処理することが可能となりリアルタイム性を実現する。したがって、証券取引所や、基地局、電気メータなど、分散拠点に設置されるノード（計算機）からデータが入力される場合、異なる拠点からのデータは時刻順に入力されないため、ナイーブな方式では、データ入力時に時刻順にソートし、ストリームデータ処理可能にする。しかし分散拠点のノードが多数存在する場合や、各拠点が地理的に離れている場合には、データ入力時の時刻順ソートにより、メモリコストや処理のレイテンシが増大するため、時刻順でない入力データへの対策として、特許文献１、特許文献２、特許文献３、特許文献４、非特許文献２、非特許文献３が開示されている。なお、メモリコストは、処理待ち等でデータを保持するために必要となる、計算機に搭載するメモリの使用量を指す。また、処理のレイテンシは、ストリームデータの処理を行う計算機へストリームデータを入力してからデータを出力するまでの遅延時間である。
【先行技術文献】
【特許文献】
【０００６】
【特許文献１】米国特許出願公開第２００８／００７２２２１号明細書
【特許文献２】米国特許出願公開第２００９／０１７２０５８号明細書
【特許文献３】米国特許出願公開第２００９／０１７２０５９号明細書
【特許文献４】米国特許出願公開第２０１０／０１０６９４６号明細書
【非特許文献】
【０００７】
【非特許文献１】Ｒ．Ｍｏｔｗａｎｉ，Ｊ．Ｗｉｄｏｍ，Ａ．Ａｒａｓｕ，Ｂ．Ｂａｂｃｏｃｋ，Ｓ．  Ｂａｂｕ，Ｍ．Ｄａｔａｒ，Ｇ．Ｍａｎｋｕ，Ｃ．Ｏｌｓｔｏｎ，Ｊ．Ｒｏｓｅｎｓｔｅｉｎ  ａｎｄ  Ｒ．  Ｖａｒｍａ著："Ｑｕｅｒｙ  Ｐｒｏｃｅｓｓｉｎｇ，Ｒｅｓｏｕｒｃｅ  Ｍａｎａｇｅｍｅｎｔ，ａｎｄ  Ａｐｐｒｏｘｉｍａｔｉｏｎ  ｉｎ  ａ  Ｄａｔａ  Ｓｔｒｅａｍ  Ｍａｎａｇｅｍｅｎｔ  Ｓｙｓｔｅｍ" ，Ｉｎ  Ｐｒｏｃ．ｏｆ  ｔｈｅ  ２００３  Ｃｏｎｆ．ｏｎ  Ｉｎｎｏｖａｔｉｖｅ  Ｄａｔａ  Ｓｙｓｔｅｍｓ  Ｒｅｓｅａｒｃｈ  （ＣＩＤＲ），Ｊａｎｕａｒｙ  ２００３
【非特許文献２】Ｊ．Ｌｉ，Ｋ．Ｔｕｆｔｅ，Ｖ．Ｓｈｋａｐｅｎｙｕｋ，Ｖ．Ｐａｐａｄｉｍｏｓ，Ｔ．Ｊｏｈｎｓｏｎ，Ｄ．Ｍａｉｅｒ著：  "Ｏｕｔ−ｏｆ−Ｏｒｄｅｒ  Ｐｒｏｃｅｓｓｉｎｇ：  ａ  Ｎｅw  Ａｒｃｈｉｔｅｃｔｕｒｅ  ｆｏｒ  Ｈｉｇｈ−Ｐｅｒｆｏｒｍａｎｃｅ  Ｓｔｒｅａｍ  Ｓｙｓｔｅｍｓ" ，Ｉｎ  Ｐｒｏｃ．ｏｆ  ｔｈｅ  ＶＬＤＢ  Ｅｎｄｏｗｍｅｎｔ，２００８
【非特許文献３】Ｂ．Ｂａｂｃｏｃｋ，Ｓ．Ｂａｂｕ，Ｍ．Ｄａｔａｒ，Ｒ．Ｍｏｔｗａｎｉ，ａｎｄ  Ｄ．Ｔｈｏｍａｓ著： "Ｏｐｅｒａｔｏｒ  Ｓｃｈｅｄｕｌｉｎｇ  ｉｎ  Ｄａｔａ  Ｓｔｒｅａｍ  Ｓｙｓｔｅｍｓ" ，２００５
【発明の概要】
【発明が解決しようとする課題】
【０００８】
上記特許文献２、特許文献３では、時刻順でない入力データの集計処理において、時刻が遅れた入力データの到着を必ずしも待たず、処理結果を近似解として算出することで、メモリコストやレイテンシの増大を抑える。しかし、近似解を用いた処理だけではデータまたは処理の整合性が保たれないため、これら特許文献２、３を適用可能な業務などの範囲は限定される。【０００９】
また、上記特許文献１では、一定時間以上データが到着しなかった場合に、処理を先に進めてよい合図をストリーム処理システムに送ることで、メモリコストや処理のレイテンシの増大を抑える。しかし、あらかじめ指定された時間よりも遅れたデータは破棄されてしまうため、処理の整合性が保たれない、という問題があった。【００１０】
上記非特許文献２では、ストリームに対して時刻順でない入力を許可する。そして、明示的に時刻を進めるために、制御パケットを送り、制御パケットがオペレータに入力されると、そのオペレータでは制御パケットの時刻までデータを処理する。この非特許文献２の処理では制御パケットを頻繁に送信する場合、その制御パケットの処理のために計算機の処理能力が劣化する、という問題があった。また、非特許文献２の処理では制御パケット送信の間隔を広げた場合には、各オペレータで制御パケットを待って処理を実行するため、処理のレイテンシやメモリコストが大きくなる、という問題があった。【００１１】
したがって、上記公知例では、処理の整合性や、レイテンシやメモリコストなど性能に関して十分とは言えない。【００１２】
そこで本発明では、処理の整合性を保ちつつ、処理のレイテンシ増大やメモリコストの増大を抑えることを課題とする。
【課題を解決するための手段】
【００１３】
プロセッサとメモリを備えたストリームデータ処理装置で、時刻が含まれた入力データで構成されるストリームデータを受信し、予め登録されたクエリに従って前記ストリームデータの処理を実行するストリームデータ処理方法であって、前記ストリームデータ処理装置は、前記ストリームデータを構成する複数の入力データを受信するデータ入力部と、前記入力データを時刻順に処理する前記入力データの項目をデータセットとして指定する第１のキーと、前記ストリームデータの定義と、前記クエリの定義を受け付けて前記入力データを処理するオペレータを生成するクエリ登録部と、前記データセット毎に前記入力データを処理するオペレータを決定し、当該オペレータで前記入力データを処理した結果を出力するデータ実行部と、を有し、前記ストリームデータ処理方法は、前記クエリ登録部が、前記第１のキーとクエリの定義及びストリームデータの定義を受け付けて、前記入力データに含まれる項目のうち、時刻順に処理するデータセットを設定する第１のステップと、前記データ入力部が、前記入力データを受信して、前記第１のキーで指定された項目で、前記入力データを前記データセット毎に分類し、前記データセット毎に前記入力データを時刻順にソートして入力ストリームを生成する第２のステップと、前記データ実行部が、前記入力ストリームを受け付けて、当該入力ストリームに含まれる入力データを、前記データセット毎に前記オペレータで処理する第３のステップと、前記オペレータが処理した結果を単一の出力ストリームとして生成する第４のステップと、を含む。
【発明の効果】
【００１４】
時刻順でない入力データを扱うストリームデータ処理において、処理の整合性を保ちながらも、処理の低レイテンシ化、低メモリコスト化を実現することができる。
【図面の簡単な説明】
【００１５】
【図１】本発明の第１の実施形態を示し、計算機システムの一例を示すブロック図である。
【図２Ａ】本発明の第１の実施形態を示し、ストリームデータ処理サーバの入出力の関係を示すブロック図である。
【図２Ｂ】本発明の第１の実施形態を示し、ストリームデータ処理サーバのデータ実行部の詳細なブロック図である。
【図３Ａ】本発明の第１の実施形態を示し、ストリーム定義の一例を示す説明図である。
【図３Ｂ】本発明の第１の実施形態を示し、クエリ定義の一例を示す説明図である。
【図３Ｃ】本発明の第１の実施形態を示し、データセットキーの一例を示す説明図である。
【図４】本発明の第１の実施形態を示し、データセット変換テーブルの一例を示す説明図である。
【図５】本発明の第１の実施形態を示し、実行領域名参照テーブルの一例を示す説明図である。
【図６】本発明の第１の実施形態を示し、実行木の一例を示すブロック図である。
【図７】本発明の第１の実施形態を示し、入力データの一例を示す説明図である。
【図８】本発明の第１の実施形態を示し、入力データ格納領域の一例を示す説明図である。
【図９】本発明の第１の実施形態を示し、入力ストリームの一例を示す説明図である。
【図１０Ａ】本発明の第１の実施形態を示し、データ実行部の実行領域Ａの一例を示すブロック図である。
【図１０Ｂ】本発明の第１の実施形態を示し、データ実行部の実行領域Ｂの一例を示すブロック図である。
【図１１】本発明の第１の実施形態を示し、実行データ、実行オペレータの一例を示す説明図である。
【図１２】本発明の第１の実施形態を示し、出力データと出力ストリームの一例を示す説明図である。
【図１３】本発明の第１の実施形態を示し、クエリ登録部の処理の一例を示すフローチャートである。
【図１４Ａ】本発明の第１の実施形態を示し、データ入力部の処理の一例を示すフローチャートの前半部である。
【図１４Ｂ】本発明の第１の実施形態を示し、データ入力部の処理の一例を示すフローチャートの後半部である。
【図１５Ａ】本発明の第１の実施形態を示し、データ実行部の処理の一例を示すフローチャートの前半部である。
【図１５Ｂ】本発明の第１の実施形態を示し、データ実行部の処理の一例を示すフローチャートの後半部である。
【図１６Ａ】本発明の第１の実施形態を示し、データ入力部とデータ実行部の処理の一例を示すタイムチャートの前半部である。
【図１６Ｂ】本発明の第１の実施形態を示し、データ入力部とデータ実行部の処理の一例を示すタイムチャートの後半部である。
【図１７Ａ】本発明の第２の実施形態を示し、ストリームデータ処理サーバの入出力の関係を示すブロック図である。
【図１７Ｂ】本発明の第２の実施形態を示し、ストリームデータ処理サーバのデータ実行部の詳細なブロック図である。
【図１８】本発明の第２の実施形態を示し、実行可能データ参照テーブルの一例を示す説明図である。
【図１９Ａ】本発明の第２の実施形態を示し、実行領域Ａの一例を示すブロック図である。
【図１９Ｂ】本発明の第２の実施形態を示し、実行領域Ｂの一例を示すブロック図である。
【図２０】本発明の第２の実施形態を示し、クエリ登録部の処理の一例を示すフローチャートである。
【図２１Ａ】本発明の第２の実施形態を示し、データ実行部の処理の一例を示すフローチャートの前半部である。
【図２１Ｂ】本発明の第２の実施形態を示し、データ実行部の処理の一例を示すフローチャートの後半部である。
【図２２Ａ】本発明の第２の実施形態を示し、データ入力部とデータ実行部の処理の一例を示すタイムチャートの前半部である。
【図２２Ｂ】本発明の第２の実施形態を示し、データ入力部とデータ実行部の処理の一例を示すタイムチャートの後半部である。
【図２３Ａ】本発明の第３の実施形態を示し、ストリームデータ処理サーバの入出力の関係を示すブロック図である。
【図２３Ｂ】本発明の第３の実施形態を示し、ストリームデータ処理サーバの詳細なブロック図である。
【図２４】本発明の第３の実施形態を示し、最大データセット数の一例を示す説明図である。
【図２５】本発明の第３の実施形態を示し、データセットとストリームの対応テーブルの一例を示す説明図である。
【図２６】本発明の第３の実施形態を示し、出力キューの一例を示す説明図である。
【図２７Ａ】本発明の第３の実施形態を示し、複製ストリームの一例を示す説明図である。
【図２７Ｂ】本発明の第３の実施形態を示し、クエリ定義の一例を示す説明図である。
【図２７Ｃ】本発明の第３の実施形態を示し、複製ストリームの一例を示す説明図である。
【図２７Ｄ】本発明の第３の実施形態を示し、クエリ定義の一例を示す説明図である。
【図２７Ｅ】本発明の第３の実施形態を示し、複製ストリームの一例を示す説明図である。
【図２７Ｆ】本発明の第３の実施形態を示し、クエリ定義の一例を示す説明図である。
【図２８】本発明の第３の実施形態を示し、クエリ登録部の処理の一例を示すフローチャートである。
【図２９】本発明の第３の実施形態を示し、データ入力部の処理の一例を示すフローチャートである。
【図３０】本発明の第３の実施形態を示し、ストリームマージ部の処理の一例を示すフローチャートを示す図である。
【図３１Ａ】本発明の第３の実施形態を示し、データ入力部とデータ実行部及びストリームマージ部の処理の一例を示すタイムチャートの前半部である。
【図３１Ｂ】本発明の第３の実施形態を示し、データ入力部とデータ実行部及びストリームマージ部の処理の一例を示すタイムチャートの後半部である。
【発明を実施するための形態】
【００１６】
以下、本発明の一実施形態について添付図面を用いて説明する。
【００１７】
＜第一実施形態＞
図１は本発明の第一の実施形態における計算機システムの一構成を示すブロック図である。送信サーバ１０１〜１０３はネットワーク１０４を介して、ストリームデータ処理システムを実行するストリームデータ処理サーバ１０８に接続されている。登録サーバ１０５はネットワーク１０７を介して、ストリームデータ処理サーバ１０８に接続されている。受信サーバ１１７はネットワーク１１６を介して、ストリームデータ処理サーバ１０８に接続されている。ネットワーク１０４、１０７、１１６は、イーサネット（登録商標）、光ファイバなどで接続されるローカルエリアネットワーク（ＬＡＮ）、もしくはＬＡＮよりも低速なインターネットを含んだワイドエリアネットワーク（ＷＡＮ）でも差し支えない。また、ストリームデータ処理サーバ１０８、送信サーバ１０１〜１０３、登録サーバ１０５、受信サーバ１１７はパーソナルコンピュータ（ＰＣ）や、ブレード型の計算機システムなどの任意のコンピュータシステムで構成することができる。
【００１８】
ストリームデータ処理サーバ１０８は、インタフェース部を構成するＩ／Ｏインタフェース１１５、処理部を構成する中央処理部（ＣＰＵ）１１３、記憶部であるメモリ１０９が、バスで結合された計算機である。
【００１９】
ストリームデータ処理サーバ１０８は、Ｉ／Ｏインタフェース１１５を介してネットワーク１０４、１０７、１１６にアクセスする。また、ＣＰＵ１１３（あるいはプロセッサ１１３でもよい）、ストリームデータ処理の結果、処理の中間結果、あるいはシステム動作に必要な設定データを不揮発性のストレージに格納する場合には、それぞれ記憶部であるストレージ装置１１４を用いることができる。ストレージ装置１１４は、Ｉ／Ｏインタフェース１１５を介して直接接続されるか、もしくはＩ／Ｏインタフェース１１５よりネットワークを介して接続される。そしてメモリ１０９には、ストリームデータ処理を構成するモジュールとして、クエリ登録部１１１、データ入力部１１０、データ実行部１１２が格納されている。各モジュールの動作については後述する。
【００２０】
以下、第一の実施形態について図面を用いて説明する。第一実施形態では、特許文献４に開示されているオペレータスケジューリングを基にした、本発明の実現方法を示す。
【００２１】
図２Ａ、図２Ｂは、第一実施形態のストリームデータ処理サーバ１０８の構成を示すブロック図である。なお、図２Ａは、ストリームデータ処理サーバ１０８の入出力の関係を示す計算機システムのブロック図である。また、図２Ｂは、ストリームデータ処理サーバ１０８のデータ実行部１１２の詳細なブロック図である。
【００２２】
まずユーザ２０４によって記述された、ストリーム及びクエリ定義２０６及び、データセットキー２０５を含む設定データは、登録サーバ１０５に格納され、登録サーバ１０５から、ストリームデータ処理サーバ１０８に送信される。そして、ストリームデータ処理サーバ１０８では、上記設定データを受信した後、クエリ登録部１１１のデータセットキー読込み部２１１において、データセットキー２０５からデータセット変換テーブル２１０及び、実行領域名参照テーブル２１５を生成する。またコンパイル部２１２において、ストリーム及びクエリ定義２０６をコンパイルし、実行木２２８を生成する。
【００２３】
ストリームデータ処理サーバ１０８でデータセット変換テーブル２１０及び、実行領域名参照テーブル２１５、実行木２２８を生成した後、送信サーバ１０１〜１０３は、入力データ２０１〜２０３をストリームデータ処理サーバ１０８に送信し続ける。
【００２４】
そしてストリームデータ処理サーバ１０８では、データ入力部１１０の入力データ受信部２０７が入力データ２０１〜２０３を受信し、受信した入力データを入力データ格納領域２０８に格納する。なお、入力データ格納領域２０８は、受信した入力データ２０１〜２０３を一時的に保持するキューで構成される。そして部分的時刻順ソート処理部２０９が、入力データ格納領域２０８に格納した入力データ２０１〜２０３を、データセット変換テーブル２１０を用いて部分的に時刻順にソートし、ソートした入力データ２１３を入力ストリーム２１４に格納する。データ入力部１１０は入力ストリーム２１４をデータ実行部１１２へ出力する。
【００２５】
そして、データ実行部１１２の実行順序決定部２１７では、入力ストリーム２１４に格納された入力データ２１３を取り出す。実行順序決定部２１７は、実行領域名参照テーブル２１５を参照し、実行領域２１８、２１９が存在する場合には、実行領域名２１６を抽出し、実行領域が存在しない場合には、新たな実行領域を生成する。なお実行領域２１８、２１９は、ストリームデータ格納キュー２２１、発火時刻参照テーブル２２２、実行状態２２３から構成される。
【００２６】
実行順序決定部２１７では、抽出した実行領域名２１６が指す実行領域２１８のストリームデータ格納キュー２２１に入力データ２１３を格納し、該実行領域２１８の発火時刻参照テーブル２２２を用いて、実行データ２２４と、実行オペレータ２２６を抽出する。
【００２７】
データ実行部１１２のオペレータ処理部２２７では、実行順序決定部２１７が抽出した実行データ２２４と、実行オペレータ２２６から実行木２２８を用いて所定の処理を実行する。オペレータ処理部２２７は処理を実行する際に、実行領域名２１６が指す実行領域２１８の実行状態２２３を用いる。
【００２８】
そしてオペレータ処理部２２７の実行による出力データ２２９は、データ出力部２３０により出力ストリーム２３１に格納する。出力ストリーム２３１に格納されたデータは、受信サーバ１１７が受け取る。また、出力ストリーム２３１に格納されたデータを、さらにオペレータ処理する場合には入力データ受信部２０７が受信する。
続いて、第一実施形態の動作の詳細を説明する。
【００２９】
図３Ａ〜図３Ｃは、ストリーム及びクエリ定義２０６に含まれるストリーム定義３０１、クエリ定義３０２と、データセットキー２０５の一例を示す。図３Ａに示す３０１は、電力の「メータ」、「使用電力」、メータの「設置場所」をカラム（またはデータ項目）とする電気メータストリームの定義である。このストリーム定義３０１は、ストリームデータを構成する入力データ２０１〜２０３に含まれる複数の項目を、ストリームデータ処理サーバ１０８で識別するための定義である。なお、入力データ２０１〜２０３には時刻（またはタイムスタンプ）が含まれるが、本実施形態では、入力データ２０１〜２０３に時刻情報が付与されていることが前提であるため、ストリーム定義３０１では定義を省略した。
【００３０】
図３Ｂに示す３０２は、電気メータストリーム定義３０１を入力し、設置場所ごとの１０分毎の合計使用電力を出力する使用電力集計クエリの定義である。このクエリ定義３０２は、入力データ２０１〜２０３を処理するオペレータを決定するための定義である。
【００３１】
そしてこのようなストリーム定義３０１と、クエリ定義３０２に対し、図３Ｃに示すデータセットキー２０５（３０３，３０４）が登録サーバ１０５でユーザにより指定される。
【００３２】
データセットキー２０５は、入力ストリーム用データセットキー（第１のキー）３０４と、入力データ用データセットキー(第２のキー)３０３から構成される。本第一実施形態では、入力データ用データセットキー３０３として入力データのカラムのうち「メータ」が設定され、入力ストリーム用データセットキー３０４として入力データのカラムのうち「設置場所」が設定された例を示す。
【００３３】
入力データ用データセットキー３０３は、指定したカラムの値が同一の入力データが時刻順に整列されて、ストリームデータ処理サーバ１０８のデータ入力部１１０へ入力されることを示す。つまり入力データは「メータ」毎に時刻順で整列された状態で入力されることを示す。
【００３４】
入力ストリーム用データセットキー３０４は、指定したカラムの値が同一の入力データを、クエリで処理するグループと扱い、データ実行部１１２の入力ストリーム２１４において時刻順に処理されることを示す。つまり入力データは「設置場所」毎に分類されてデータセットを構成し、データセット毎の入力データが時刻順で処理されることを示す。また、入力ストリーム用データセットキー３０４は、入力データの「設置場所」の種類に応じた数の実行領域をデータ実行部１１２で生成することを示す。換言すれば、入力データの「設置場所」の種類毎にデータセットが構成される。
【００３５】
なお入力データ用データセットキー３０３と入力ストリーム用データセットキー３０４を指定する方法は、クエリに記述してもよいし、設定ファイル等を介して指示してもよく、あるいはその他の方法でも構わない。
【００３６】
図４は、データセット変換テーブル２１０の一例を示す。データセット変換テーブル２１０は、入力データ格納領域２０８に格納された入力データについて、図３Ｃで設定された入力データ用データセットキー３０３と、入力ストリーム用データセットキー３０４の関係を設定するテーブルである。
【００３７】
図４では、入力データ用データセットキー４０１としてメータ、入力ストリーム用データセットキー４０２として設置場所が指定された例であり、図３Ｃのデータセットキーの設定から、メータ「メータ００」と設置場所「田中家」（４０３）、メータ「メータ１０」と設置場所「佐藤ビル」（４０４）、メータ「メータ１１」と設置場所「佐藤ビル」（４０５）が対応していることを示す。
【００３８】
図５は、実行領域名参照テーブル２１５の一例を示す。実行領域名参照テーブル２１５は、入力ストリーム用データセットキー値４０１と、実際に入力データの処理を行う独立した実行領域名５０２の対応関係を示す。図５では、図４で示したように、入力ストリーム用データセットキー４０２として設置場所が指定された例であり、設置場所「田中家」と実行領域名「実行領域Ａ」（５０３）、設置場所「佐藤ビル」と実行領域名「実行領域Ｂ」（５０４）がそれぞれ対応していることを示す。
【００３９】
図５の例では、図３Ｃのデータセットキーのうち、入力ストリーム用データセットキー３０４ごとに、独立した実行領域を生成する例を示し、入力データ格納領域２０８内に「田中家」と「佐藤ビル」の２種類の入力ストリーム用データセットキー値４０２が存在し、実行領域名参照テーブル２１５は、各入力ストリーム用データセットキー値４０２毎に独立した実行領域Ａ、Ｂ（５０３、５０４）を生成することを示唆する。
【００４０】
図６は、実行木２２８の一例を示す図である。図６の実行木は、使用電力集計のクエリ定義３０２をコンパイルしてクエリ登録部１１１で生成された実行木を表し、オペレータＲＡＮＧＥ ６０１、オペレータＧＲＯＵＰ  ＢＹ ６０２と、オペレータＩＳＴＲＥＡＭ ６０３で構成される。図６に示す実行木は、データ実行部１１２のオペレータ処理部２２７において、ＲＡＮＧＥ６０１、ＧＲＯＵＰ  ＢＹ６０２、ＩＳＴＲＥＡＭ６０３の順に実行することを示す。
【００４１】
図７は、送信サーバ１０１〜１０３が送信する入力データの一例を示す図である。図７は、図３Ａに示した電気メータストリーム定義３０１への入力データを示す。各入力データの到着順７０１、入力データに付加された時刻７０２と、電気メータストリーム定義３０１の項目である、メータ７０３、使用電力７０４、設置場所７０５の値を示す。
【００４２】
図７では、入力データ７０６、入力データ７０７、入力データ７０８の順でストリームデータ処理サーバ１０８が受信し、入力データ７０６は時刻「７：５９」、メータ「メータ１０」、使用電力「５０Ｗ／分」、設置場所「佐藤ビル」であり、入力データ７０７は時刻「８：００」、メータ「メータ０１」、使用電力「１００Ｗ／分」、設置場所「田中家」であり、入力データ７０８は時刻「７：５９」、メータ「メータ１１」、使用電力「２００Ｗ／分」、設置場所「佐藤ビル」となっている。
【００４３】
図８は、入力データ格納領域２０８の一例を示す図である。入力データ格納領域２０８は、送信サーバ１０１〜１０３から受信した入力データを、図４に示した入力データ用データセットキー値４０１ごとに、キュー８０５〜８０７に格納する。図８は、図４で示したように、入力データ用データセットキー値４０１として「メータ」が指定された場合に、電気メータストリーム定義３０１の入力データのうち「メータ」を格納する、入力データ格納領域２０８を示す。
【００４４】
図８において、８０５は「メータ１０」（８０２）の入力データ７０６を格納するキュー、８０６は「メータ０１」（８０３）の入力データ７０７を格納するキュー、８０７は「メータ１１」（８０４）の入力データ７０８を格納するキューである。すなわち、データ入力部１１０は、入力データ用データセットキー値４０１ごとに入力データを格納するキューをメモリ１０９上に生成して、入力データ２０１〜２０３（７０６，７０７，７０８）を格納していく。
【００４５】
図９は、入力ストリーム２１４の一例を示す図である。入力ストリーム２１４は、オペレータ処理を実行する入力データを格納する。本発明では、従来例とは異なり、入力ストリーム２１４内の全ての入力データが時刻順に並んでいる必要はなく、入力データ用データセットキー値４０１が同一のデータについて時刻順に並んでいればよい。２１４は、図３Ａに示した電気メータストリームの定義（３０１）に従ってデータ入力部１１０が生成した入力ストリームの例である。
【００４６】
図１０Ａ、図１０Ｂは、独立して生成されるオペレータ処理の実行領域２１８，２１９の一例を示す図である。実行領域は、入力ストリーム用データセットキーの値４０２ごとにデータ実行部１１２が生成する。図１０Ａでは、入力ストリーム用データセットキー値４０２が設置場所「田中家」の実行領域として、実行領域Ａ（２１８）が生成され、図１０Ｂでは、入力ストリーム用データセットキー値４０２が設置場所「佐藤ビル」の実行領域として、実行領域Ｂ（２１９）が生成された例を示す。各実行領域２１８、２１９には、入力ストリーム２１４を格納するストリームデータ格納キュー１００２、１０１５、入力ストリーム２１４に対してオペレータ処理を開始する時刻を設定した発火時刻参照テーブル１００４、１０１７と、オペレータの状態を示す実行状態１００８、１０２１を含む。なお、実行状態１００８、１０２１の総称は２２３とする。また、実行状態２２３は、例えば、オペレータの処理結果を格納し、当該格納した値でオペレータの状態を表すことができる。
【００４７】
ストリームデータ格納キュー１０００、１０１５は、入力ストリーム２１４の入力ストリーム用データセットキー値４０２を持つ入力データ２１３を格納するキューである。なお、ストリームデータ格納キュー１０００、１０１５の総称を符号２２１で表す。
【００４８】
発火時刻参照テーブル１００４，１０１７は、ストリームデータ格納キュー１００２，１０１５に格納された入力データの処理について、処理を実行するオペレータ１００５、１０１８と、当該オペレータが実行される時刻である発火時刻１００６、１０１９を格納する。
【００４９】
例えば、実行領域Ａ（２１８）におけるオペレータＲＡＮＧＥの発火時刻１００６は、オペレータＲＡＮＧＥ内の最古データの時刻が「７：５１」である場合、図３Ｂの使用電力集計クエリ定義３０２に示すように、集計期間が１０分間ウィンドウであるため、前回の実行時刻に１０分を加えた「７：５１＋１０分」により、次回の発火時刻は「８：０１」（１００７）となる。同様に実行領域ＢにおけるオペレータＲＡＮＧＥの発火時刻は、オペレータＲＡＮＧＥ内の最古データの時刻が「７：５０」である場合、同じく１０分間ウィンドウであるため、７：５０＋１０分により、次回の発火時刻は「８：００」（１０２９）となる。これらの発火時刻は、実行順序決定部２１７が設定することができる。
【００５０】
また実行状態１００８、１０２１は、ストリームデータ格納キュー１００２，１０１５に格納された入力データの処理において、各オペレータの処理に用いる状態を示す。例えば、オペレータＲＡＮＧＥの実行状態である、実行状態Ａ１（１０１１）、実行状態Ｂ１（１０２４）は、１０分間ウィンドウであるため、各々実行領域に、現在時刻から１０分以内の時刻の入力データを格納する。また、オペレータＧＲＯＵＰ  ＢＹの実行状態を示す実行状態Ａ２（１０１２）、実行状態Ｂ２ （１０２５）は、現在時刻から１０分以内の時刻の入力データの集計値を格納する。
【００５１】
図１１は、実行データ２２４と、実行オペレータ２２６の一例を示す。実行データ２２４はオペレータ処理部２２７で実行するデータであり、実行オペレータ２２６はオペレータ処理部２２７で実行データ２２４に対する処理を実行するオペレータである。図１１では実行データ２２４として、図７に示した入力データ７０６と、実行オペレータ２２６として図６に示したオペレータＲＡＮＧＥ６０１を例示する。
【００５２】
図１２は、出力データ２２９と出力ストリーム２３１の一例を示す図である。出力データは、受信サーバ１１７や入力データ受信部２０７において取得されるオペレータの処理結果である。また出力ストリーム２３１は出力データを格納する領域である。図中２３１は、使用電力集計クエリ定義３０２の出力データ１２０２〜１２０４を格納する出力ストリームであり、入力データ７０６、入力データ７０７、入力データ７０８の処理結果が、それぞれ出力データ１２０３、出力データ１２０４、出力データ１２０２である。なお、出力データ１２０２〜１２０４の総称を図２Ｂで示した符号２２９で表す。
【００５３】
図１３は、クエリ登録部１１１の処理の一例を示すフローチャートである。まず、前記従来例のストリームデータ処理方法と同様に、クエリ登録部１１１は、登録サーバ１０５で定義されたストリーム定義３０１、クエリ定義３０２を受信した後にコンパイルし、入力データを処理するオペレータと当該オペレータの実行木２２８（図６）を生成する。クエリ登録部１１１は、生成した実行木２２８をオペレータ処理部２２７に格納する（１３０７）。
【００５４】
そしてコンパイル後、クエリ登録部１１１は、データセットキー読込み部２１１において、データセットキー２０５の読み込みを開始する（１３０１）。データセットキー読み込み部２１１は、登録サーバ１０５からデータセットキー２０５を構成する入力データ用データセットキー３０３と、入力ストリーム用データセットキー３０４を受信する（１３０２）。
【００５５】
次に、データセットキー読み込み部２１１は、受信した入力ストリーム用データセットキー値４０２と実行領域名５０２の対応関係を示す実行領域名参照テーブル２１５（図５）を生成する（１３０３）。生成された実行領域名参照テーブル２１５は、図５に示したエントリ５０３、５０４のようなデータが存在してもよいし、データが存在しない空のテーブルであってもよい。
【００５６】
そして、クエリ登録部１１１は、入力データ用データセットキー３０３と入力ストリーム用データセットキー３０４が一致した場合には（１３０４）、データセットキー読込み部を終了する（１３０６）。
【００５７】
一方、ステップ１３０４において、入力データ用データセットキー３０３と入力ストリーム用データセットキー３０４が一致しなかった場合には、入力データ用データセットキー値４０１と入力ストリーム用データセットキー値４０２の対応関係を示すデータセット変換テーブル２１０（図４）を生成する（１３０５）。生成したデータセット変換テーブル２１０は、図４に示したエントリ４０３〜４０５のようなデータが存在してもよいし、データが存在しない空のテーブルであってもよい。  
【００５８】
そして、クエリ登録部１１１は、データセット変換テーブル２１０を生成した後、データセット読み込み部を終了する（１３０６）。
【００５９】
上記処理により、クエリ登録部１１１は、登録サーバ１０５で定義されたストリーム定義３０１、クエリ定義３０２をコンパイルしてオペレータと実行木２２８を生成し、登録サーバ１０５で定義されたデータセットキー２０５からデータセット変換テーブル２１０と実行領域名参照テーブル２１５を生成する。なお、この処理は、クエリ登録部１１１が、登録サーバ１０５からストリーム及びクエリ定義２０６やデータセットキー２０５を受け付けたときに実行することができる。
【００６０】
図１４Ａ、図１４Ｂは、データ入力部１１０で行われる処理の一例を示すフローチャートである。データ入力部１１０では、まず入力データ受信部２０７において（１４０１）、送信サーバ１０１〜１０３から入力データ２０１〜２０３を受信する（１４０２）。
【００６１】
そして、データ入力部１１０は、データセット変換テーブル２１０の有無を判定する（１４０３）。データ入力部１１０は、データセット変換テーブルが存在しない場合には、入力データ２０１〜２０３を入力ストリーム２１４に格納し（１４０５）、入力ストリーム２１４をデータ実行部１１２へ出力してから入力データ受信部２０７の処理を終了する（１４０６）。
【００６２】
一方、上記ステップ１４０３において、データセット変換テーブル２１０が存在する場合には、データ入力部１１０が以下の処理を実行する。まずデータ入力部１１０は、データセット変換テーブル２１０に、入力データ２０１〜２０３の項目に入力データ用データセットキー値４０１が存在する場合には（１４０４）、当該入力データ用データセットキーの値４０１に従って、入力データ格納領域２０８のキューに入力データを格納し（１４０８）、入力データ受信部２０７の処理を終了する（１４０９）。
【００６３】
例えば、図７に示した入力データ７０８をデータ入力部１１０で受信したとき（到着時）の、データセット変換テーブル２１０、入力データ格納領域２０８を、図４、図８に示す。
【００６４】
今、図３Ｃの３０３のように、入力データ用データセットキーとして「メータ」が指定されており、図７に示す入力データ７０８のメータ７０３の値は「メータ１１」であることから、データ入力部１１０は、入力データ格納領域２０８のキュー８０７を生成して入力データ７０８を格納する。
【００６５】
また、上記ステップ１４０４において、データセット変換テーブル２１０に、入力データ２０１〜２０３の入力データ用データセットキー値４０１が存在しない場合には、データセット変換テーブル２１０に、データ入力部１１０が受信した入力データの入力データ用データセットキー値４０１と入力ストリーム用データセットキー値４０２を格納する。また、データ入力部１１０は、入力データ格納領域２０８に、入力データの入力データ用データセットキー値４０１に対応するキューを追加する（１４０７）。
【００６６】
そして、データ入力部１１０は、入力データを、入力データ用データセットキー値４０１に従って入力データ格納領域２０８のキューに格納し（１４０８）、入力データ受信部２０７の処理を終了する（１４０９）。
【００６７】
なお送信サーバ１０１〜１０３からの入力データ２０１〜２０３ではなく、入力データ用データセットキー値４０１、入力ストリーム用データセットキー値４０２と、時刻を持つダミーデータをデータ入力部１１０へ入力することにより、入力データ受信部２０７で、上記データ入力時と同様に、データセット変換テーブル２１０に、ダミーデータの入力データ用データセットキー値４０１と入力ストリーム用データセットキー値４０２を格納し、また入力データ格納領域２０８に、ダミーデータの入力データ用データセットキー値４０１に対応するキューを追加しても構わない。
【００６８】
また、入力データ用データセットキー値４０１、入力ストリーム用データセットキー値４０２、時刻と、終了フラグを持つダミーデータを入力し、以下のような処理を入力データ受信部２０７で実行することができる。処理の一例としては、まず入力データ受信部２０７で、終了フラグを読み取る。そして、終了フラグが所定の値の場合には、入力データ受信部２０７が、ダミーデータの入力データ用データセットキー値４０１から抽出した入力データが属するデータセットが、データセット変換テーブル２１０に存在する場合は、データセット変換テーブル２１０から、ダミーデータの入力データ用データセットキー値４０１と入力ストリーム用データセットキー値４０２を削除する。また入力データ格納領域２０８からダミーデータの入力データ用データセットキー値４０１に対応するキューを削除する。上記ダミーデータを用いることで、データセット変換テーブル２１０のエントリと、入力データ格納領域２０８のキューを削除することができる。これにより、データ入力部１１０のメモリ１０９使用量が過大になるのを抑制できる。
【００６９】
なお、入力データ格納領域２０８のキューは、図８で示した８０５〜８０７のように、データ入力部１１０がデータの到着順に生成することができる。データ入力部１１０は、入力データ用データセットキー値４０１の値（本実施形態では「メータ」）ごとにキューを生成する。図８の例では、データ入力部１１０は、「メータ１０」の入力データを格納するキュー８０５と、「メータ０１」の入力データを格納するキュー８０６と、「メータ１１」の入力データを格納するキュー８０７を生成した例を示す。各キュー８０５〜８０７にはデータ入力部１１０へ到着した順序で「メータ」毎に分類されて入力データが格納される。つまり、データ入力部１１０は、入力データ用データセットキー値４０１で指定された項目毎に、入力データを分類してキュー８０５〜８０７に格納する。
【００７０】
次に、上記ステップ１４０４において、データセット変換テーブル２１０が存在する場合には、次に部分的時刻順ソート処理部２０９の処理を開始する（１４１０）。
【００７１】
部分的時刻順ソート処理部２０９では、まず入力データ格納領域２０８において、入力ストリーム用データセットキー値４０２が等しいデータが格納されたキューの先頭データの時刻を比較する（１４１１）。この処理は、入力データ用データセットキー値４０１が異なり入力ストリーム用データセットキー値４０２が等しい入力データ格納領域２０８のキューに格納された先頭の入力データの時刻（図７の時刻７０２）を部分的時刻順ソート処理部２０９が比較する。
【００７２】
すなわち、部分的時刻順ソート処理部２０９は、図７に示す入力データを、入力データ格納領域２０８のキュー８０５〜８０７へ格納してから、第１のキーである入力ストリーム用データセットキー値４０２＝「設置場所」が同一で、かつ、第２のキーである入力データ用データセットキー値４０１＝「メータ」が異なる入力データの時刻７０２を比較する。
【００７３】
次に、部分的時刻順ソート処理部２０９は、上記比較の結果、入力データ格納領域２０８内で入力ストリーム用データセットキー値４０２が等しい入力データのうち、最も時刻が古いデータ（以下、最古データとする）が存在するか否かを判定する（１４１２）。
【００７４】
そして最古データが存在する場合には（１４１２）、最古データを入力データ格納領域２０８のキュー８０５〜８０７から取得し、入力ストリーム２１４に格納する（１４１３）。そしてステップ１４１１〜１４１３の処理を、最古データが存在する限り繰り返し、最古データがなくなるとステップ１４０２に進んで、送信サーバから新たな入力データを受信する。
【００７５】
例えば、入力データ７０８をデータ入力部１１０で受信した後の、入力ストリーム２１４を図９に示す。入力ストリーム用データセットキー３０４しては図３Ｃで示したようにメータの「設置場所」が指定されており、図８で示したように、入力データ７０８のメータの設置場所は「佐藤ビル」であることから、設置場所が「佐藤ビル」の入力データが格納されたキュー８０５、８０７の先頭データ７０６、７０８の時刻を比較する。そして、キュー８０５のデータ７０６よりも時刻が古いデータがキュー８０７に存在しないため、最古データをデータ７０６とし、入力ストリーム２１４に格納する。また同じくデータ７０８よりも時刻が古いデータがキュー８０５に存在しないため、最古データをデータ７０８とし、入力ストリーム２１４に格納する。この結果、図９で示すように、部分的時刻順ソート処理部２０９は、入力ストリーム用データセットキー値４０２が「佐藤ビル」に等しい入力データ７０６、７０８を時刻の古い順にソートした入力ストリーム２１４を生成し、データ実行部１１２へ出力する。
【００７６】
上記処理によって、部分的時刻順ソート処理部２０９は、図７の入力データから第１のキーである入力ストリーム用データセットキー値４０２で指定された「設置場所」が等しく、第２のキーである入力データ用データセットキー値４０１で指定された「メータ」が異なる入力データを時刻７０２の古い順にソートして入力ストリーム２１４を生成して、入力ストリーム２１４をデータ実行部１１２へ出力する。したがって、部分的時刻順ソート処理部２０９は、第１のキーである入力ストリーム用データセットキー値４０２で指定された「設置場所」ごとに入力データをまとめて時刻順にソートした入力ストリーム２１４を出力することができる。
【００７７】
図１６Ａ、図１６Ｂは、入力データ７０７〜７０８がストリームデータ処理サーバ１０８に到着した時の、データ入力部１１０とデータ実行部１１２のタイムチャートを示す。
【００７８】
データ入力部１１０では、前述の図９のように、入力データ７０８を受信した時に、入力データ７０６及び、入力データ７０８を入力ストリーム２１４に格納し、データ実行部１１２へ出力する。  また入力データ７０７を受信した時に、この入力データ７０７の入力ストリーム用データセットキー値４０２である「田中家」を設置場所とするデータで、入力データ用データセットキー値が「メータ０１」以外をメータとする入力データは存在しないので、「メータ０１」のデータの中で最古データである入力データ７０７が、「田中家」の中でも最古データとなるため、入力データ７０７を入力ストリーム２１４に格納し、データ実行部１１２へ出力する。
【００７９】
図１５Ａ、図１５Ｂは、データ実行部１１２の処理の一例を示すフローチャートである。この処理は、データ入力部１１０から入力ストリーム２１４を受け付けたときに実行を開始することができる。
【００８０】
まず実行順序決定部２１７（１５０１）は、データ入力部１１０が出力した入力ストリーム２１４から入力データ２１３を取得する。そして実行順序決定部２１７は、入力データ２１３の入力ストリーム用データセットキーの値４０２から、実行領域名参照テーブル２１５を参照し、入力ストリーム用データセットキー値４０２に対応する実行領域名を抽出し、入力データ２１３が属するデータセットの実行領域として設定する。その際に、実行順序決定部２１７は、入力データ２１３について同一のデータセット（入力ストリーム２１４）内において時刻順に並んでいるか否かを検査してもよい。
【００８１】
次に、実行順序決定部２１７は、実行領域名参照テーブル２１５に入力データ２１３の入力ストリーム用データセットキー値４０２に対応する実行領域が存在するか否かを判定する（１５０３）。
【００８２】
入力ストリーム用データセットキー値４０２に対応する実行領域が存在する場合には、実行順序決定部２１７が、入力データ２１３の所属する実行領域のストリームデータ格納キュー２２１に入力データを格納する（１５０５）。
【００８３】
一方、上記ステップ１５０３の判定で、実行領域名参照テーブル２１５に、入力データ２１３の入力ストリーム用データセットキー値４０２が存在しなかった場合には、実行順序決定部２１７が当該入力ストリーム用データセットキー値４０２に対応する実行領域を生成し、実行領域名参照テーブル２１５に、当該入力ストリーム用データセットキー値４０２と、生成した実行領域名を追加する（１５０４）。そして実行順序決定部２１７が生成した実行領域を、入力データ２１３が属するデータセットの実行領域とし、当該実行領域のストリームデータ格納キュー２２１に入力データ２１３を格納する（１５０５）。なお当該実行領域の発火時刻参照テーブル２２２は、１００７のようなエントリが存在しない空のテーブルである。また当該実行領域の実行状態２２３も、１０１１、１０１２のようなエントリが存在しない空の領域である。発火時刻参照テーブル２２２、実行状態２２３はオペレータ処理部２２７が更新する。
【００８４】
例えば、図９に示した入力ストリーム２１４から入力データ７０６を取得した場合、実行順序決定部２１７は、入力データ７０６の入力ストリーム用データセットキー値４０２は「佐藤ビル」であるため、実行領域名参照テーブル２１５（図５参照）から、入力ストリーム用データセットキー値４０２＝「佐藤ビル」の実行領域として「実行領域Ｂ」を抽出し、図１０Ｂに示した実行領域Ｂ２１９のストリームデータ格納キュー１０１５に入力データ７０６を格納する。同様に、図１６Ａに示すように、入力データ７０７は、入力ストリーム用データセットキー値４０２が「田中家」であるから、図１０Ａに示した実行領域Ａ２１８のストリームデータ格納キュー１００２に格納する。また入力データ７０８は、入力ストリーム用データセットキー値４０２が「佐藤ビル」であるから、図１０Ｂの実行領域Ｂ２１９のストリームデータ格納キュー１０１５に格納する。
【００８５】
なおダミーデータを入力することにより、実行順序決定部２１７で、ダミーデータの入力ストリーム用データセットキー値４０２に対応する実行領域を生成し、実行領域名参照テーブル２１５に、ダミーデータの入力ストリーム用データセットキー値４０２と、生成した実行領域名を追加してもよい。また、終了フラグを持つダミーデータを入力することにより、以下のような処理を実行順序決定部２１７で実行することができる。処理の一例としては、実行順序決定部２１７が、まず終了フラグを読み取る。そして終了フラグが所定の値のときには、ダミーデータの入力ストリーム用データセットキー値４０２に対応する実行領域が存在する場合には、該実行領域を削除し、実行領域名参照テーブル２１５から入力ストリーム用データセットキー値４０２と実行領域名５０２を削除する。
【００８６】
次に、実行順序決定部２１７は、入力データ２１３が属するデータセットの実行領域のストリームデータ格納キュー２２１を参照し、先頭データの時刻（複数の入力ストリームのデータを処理するクエリの場合には、複数のストリームデータ格納キューの先頭データの時刻）と、発火時刻参照テーブル２２２の各オペレータの発火時刻を比較し、最も時刻が古いデータがストリームデータ格納キュー２２１に存在する場合には（１５０６）、当該データを実行データとする。実行順序決定部２１７は、また、実行データ２２４がストリームデータ格納キュー２２１の先頭データの場合には、実行木２２８の最初のオペレータを実行オペレータとして設定する。また、実行順序決定部２１７は、現在時刻がオペレータの発火時刻に対応するデータである場合には、当該発火時刻に対応するオペレータを実行オペレータとして設定（１５０７）して、実行順序決定部２１７の処理を終了する（１５０８）。
【００８７】
上記ステップ１５０６において、最も時刻が古いデータが存在しなかった場合には、実行順序決定部２１７は、再びステップ１５０２の処理に戻って入力ストリーム２１４から次の入力データ２１３を取得し、上記と同様の処理を繰り返す。
【００８８】
例えば、実行領域Ｂ２１９のストリームデータ格納キュー１０１６に、実行順序決定部２１７が入力データ７０６を格納した場合には、入力データ７０６の時刻「７：５９」と、発火時刻参照テーブル２２２に格納されているオペレータＲＡＮＧＥの発火時刻「８：００」（１０２０）を実行順序決定部２１７が比較し、入力データ７０６の時刻「７：５９」が古いため、実行データ２２４を入力データ７０６とし、実行木２２８（図６）の最初のオペレータ６０１を、実行オペレータ２２６としてオペレータ処理部２２７に指令する。
【００８９】
そしてオペレータ処理部２２７で、実行データ２２４の入力ストリーム用データセットキー値４０２に対応する実行領域上の実行状態２２３を用いて、実行データ２２４を実行オペレータ２２６で処理する（１５０９）。
【００９０】
次に、データ実行部１１２は、実行オペレータ２２６の処理結果を出力するデータ出力部２３０の処理を行う（１５１０）。まず、データ出力部２３０は、実行オペレータ２２６の処理結果について出力データが存在するか否かを判定する（１５１１）。出力データが存在しない場合には、ステップ１５１３へ進んでデータ出力部２３０の処理を終了する。一方、出力データが存在する場合には、データ出力部２３０はステップ１５１２を実施する。
【００９１】
実行オペレータ２２６の処理結果を、受信サーバ１１７や入力データ受信部２０７で出力データとして取得する場合には、データ出力部２３０は、実行オペレータ２２６の処理結果である出力データ２２９を実行領域ごとではなく、単一のストリームにマージする（１５１２）。データ出力部２３０は、複数の出力データ２２９をマージして出力ストリーム２３１として出力する。データ実行部１１２は、データ出力部２３０の処理が終了すると（１５１３）、実行順序決定部２１７の処理を再開する（１５１４）。
【００９２】
ステップ１５１５では、実行順序決定部２１７が実行木２２８に次のオペレータが存在するか否かを判定する（１５１５）。実行木２２８に次のオペレータが存在する場合には、ステップ１５１６へ進み、オペレータが存在しない場合にはステップ１５０６へ戻って上記処理を繰り返す。ステップ１５１６では、実行順序決定部２１７が実行木２２８の次のオペレータを実行オペレータ２２６として決定し、ステップ１５０８へ戻り、上記処理を繰り返す。
【００９３】
つまり、データ実行部１１２では、実行木２２８に次のオペレータが存在する限り（１５１５）、次のオペレータを実行オペレータ２２６とし、処理が終わった実行データ２２４と同一データセット（入力ストリーム用データセットキーの値４０２が同一）かつ同一時刻の次のデータを実行データ２２４とし、実行データ２２４の入力ストリーム用データセットキー値４０２に対応する実行領域上の実行状態２２３を用いて処理を続ける。
【００９４】
データ実行部１１２は、ステップ１５１５において実行木２２８に次のオペレータがなくなった場合には、ステップ１５０６において実行可能なデータを抽出し、前記したように処理する。なお１５０６において実行可能なデータを抽出できなかった場合には、１５０２において入力ストリームから入力データを取得し、前述したように動作する。
【００９５】
例えば、図１６Ｂに示すように、入力データ７０６を実行データ２２４、オペレータＲＡＮＧＥ（６０１）を実行オペレータ２２６としてオペレータ処理した後に、その処理結果の１６０３を実行データ２２４とし、次のオペレータＧＲＯＵＰ  ＢＹ（６０２）を実行オペレータ２２６としてオペレータ処理を行う。データ実行部１１２では、さらにオペレータＧＲＯＵＰ  ＢＹ（６０２）の処理結果の１６０４を実行データ２２４とし、次のオペレータＩＳＴＲＥＡＭ（６０３）を実行オペレータ２２６としてオペレータ処理を行い、その処理結果１２０３を受信サーバ１１７へ送信するため、出力データとして出力ストリーム２３１に格納する。また実行木２２８（図６）に次のオペレータが存在しないため、続いて入力データ７０８を実行データ２２４としてオペレータ処理を実行する。
【００９６】
以上で記述した第一実施形態は、特許文献４に開示されているオペレータスケジューリングを基にした、本発明の一つの実現方法であり、他にも様々な実現方法がある。例えば、入力ストリーム用データセットキー値４０２ごとに実行領域２１８、２１９を分けずに、発火時刻参照テーブル１００４、１０１７と、ストリームデータ格納キュー１００２、１０１５を入力ストリーム用データセットキー値４０２ごとに分け、実行状態１００８、１０２１を入力ストリーム用データセットキー値ごとに分けず、クエリで定義することによりオペレータ毎に実行状態を分ける方法も可能である。
【００９７】
なお、上記第一実施形態では、実行順序決定部２１７において、発火時刻参照テーブル２２２、ストリームデータ格納キュー２２１を参照する点で、上記特許文献４と共通しているが、実行領域名参照テーブル２１５及び、実行領域名２１６、実行領域２１８、２１９を参照する点で特許文献４とは異なり、本発明に固有の構成である。また第一実施形態では、特許文献４とは異なり、データセットキー読込み部２１１、データセットキー２０５、部分的時刻順ソート処理部２０９、データセット変換テーブル２１０、データ出力部２３０を含む点が、本発明に固有の構成である。
【００９８】
以上のように、本発明では、時刻を含む入力データ２０１〜２０３（以下、単に入力データとする）を処理するストリームデータ処理サーバ１０８がデータセットキー２０５を受け付けて、データセット変換テーブル２１０を生成する。ここで、データセット変換テーブル２１０は、同一の実行領域で処理する入力データの種類（グループ）を定義する入力ストリーム用データセットキー４０２（第１のキー）と、時刻順に整列された状態で入力される入力データの項目を定義する入力データ用データセットキー４０１（第２のキー）の２つのキーを有する。また、クエリ登録部１１１では、ストリーム及びクエリの定義２０６を受け付けて実行木２２８を生成し、データ実行部１１２に出力する。
【００９９】
ストリームデータ処理サーバ１０８のデータ入力部１１０は、入力データを入力データ用データセットキー４０１毎にまとめてから、入力ストリーム用データセットキー４０２毎に時刻順にソートして入力ストリーム２１４を生成する。
【０１００】
そして、データ実行部１１２では、データセット変換テーブル２１０の入力ストリーム用データセットキー４０２毎に、メモリ１０９上に入力データを処理する領域として実行領域２１８，２１９を設定する。換言すれば、入力ストリーム用データセットキー４０２の項目で分類された入力データのグループ（データセット）毎に実行領域が生成される。
【０１０１】
データ実行部１１２では、入力ストリーム２１４に含まれる入力データの種類（データセット）毎の実行領域で、入力データに対応するオペレータが決定されて、当該オペレータにより所定のクエリ処理を行い、出力ストリーム２３１を出力する。
【０１０２】
上記処理により、第１のキーである入力ストリーム用データセットキー値４０２毎に異なるメモリ１０９上の実行領域２１８、２１９でオペレータによって入力データ２１３を処理することができる。すなわち、一つのストリームデータで、第１のキーである入力ストリーム用データセットキー値４０２が同一の入力データについてのみ、時刻の順序を保ってオペレータによる処理を行うことができる。これにより、データ実行部１１２の各実行領域２１８、２１９は、第１のキーである入力ストリーム用データセットキー値４０２が異なる入力データについては、時刻が遅れた入力データの到着を待たずにオペレータによる処理を実行でき、処理の整合性を保ちながら、処理のレイテンシが増大するのを抑制できる。
【０１０３】
すなわち、データ入力部１１０では、データ実行部１１２でクエリを実行する前にデータセットごとに時刻順で入力データをソートすることにより、複数の送信サーバ１０１〜１０３から、複数の入力データを受け付けても、ひとつのサーバ内のデータセットと同様に実行順序を決定することができる。
【０１０４】
これにより、時刻順でない入力データを受け付けても前記従来例のように制御パケットで時刻を進めたり、データの破棄などの処理を行うことがなくなって、データ処理の整合性を保ちつつ、処理のレイテンシが増大するのを防ぐことができる。また、前記従来例のように制御パケットを待つ間のメモリ領域も不要となるので、メモリコストの増大を抑制できる。
【０１０５】
さらに、データ実行部１１２では、メモリ１０９上の実行領域をストリームデータ処理中に動的に生成することができる。つまり、ストリームデータ処理が開始されても、ストリームデータ処理サーバ１０８は実際に入力ストリーム用データセットキー４０２に対応する入力データを受け付けるまで、実行領域を生成しない。このため、本発明のストリームデータ処理サーバ１０８は、ストリームデータ処理に必要な実行領域のみをメモリ１０９上に確保するので、前記従来例のようにメモリコストが増大するのを抑制できる。
【０１０６】
＜第二実施形態＞
続いて本発明の第二実施形態について図面を用いて説明する。第二実施形態では、非特許文献３に開示されているラウンドロビンのオペレータスケジューリングを基にした、本発明の実現方法を示す。
【０１０７】
図１７Ａ、図１７Ｂは第二実施形態のブロック図である。なお、図１７Ａはストリームデータ処理サーバ１０８の入出力の関係を示す計算機システムのブロック図である。また、図１７Ｂは、ストリームデータ処理サーバ１０８のデータ実行部１１２の詳細なブロック図である。
【０１０８】
第二実施形態では、クエリ登録部１１１のデータセットキー読込み部２１１において、データセットキー２０５及び、コンパイル部２１２で生成された実行木２２８から、実行可能データ参照テーブル１７０１を生成する点が前記第一実施形態と相違する。またデータ入力部１１０では前記第一実施例と同様の処理が行われる。そしてデータ実行部１１２の実行順序決定部２１７は、実行可能データ参照テーブル１７０１を用いて、実行データ２２４、実行オペレータ２２６を抽出する。また、図１７Ｂで示すように、実行領域２１８Ａ、２１９Ａは、前記第一実施形態と異なり、発火時刻参照テーブルを削除した点が相違する。なお、データ実行部１１２のオペレータ処理部２２７及び、データ出力部２３０は、前記第一実施形態と同様に処理を行う。また、前記第一実施形態と同一の構成については、図２Ａ、図２Ｂと同一の符号を付した。
【０１０９】
図１８は、実行可能データ参照テーブル１７０１の一例を示す説明図である。実行可能データ参照テーブル１７０１は、クエリ登録部１１１で生成されてデータ実行部１１２で更新される。実行可能データ参照テーブル１７０１は、実行木２２８の各オペレータが、入力ストリーム用データセットキー値４０２ごとに、実行可能なデータが存在するか否かを示す。図１８では、入力ストリーム用データセットキー値４０２として「田中家」１８０５、「佐藤ビル」１８０６があり、それぞれオペレータＲＡＮＧＥ １８０２、ＧＲＯＵＰ  ＢＹ １８０３、ＩＳＴＲＥＡＭ １８０４に実行可能なデータが存在するか否かを、フラグ「○」が格納されているか否かで示す。
【０１１０】
図１９Ａ，図１９Ｂは、実行領域２１８Ａ、２１９Ａの一例を示すブロック図である。第二実施形態における実行領域２１８Ａ、２１９Ａは、前記第一実施形態の実行領域２１８、２１９とは異なり、発火時刻参照テーブル１００４、１０１７を含まず、ストリームデータ格納キュー１００２、１０１５と、実行状態１００８、１０２１を含む。
【０１１１】
図２０は、クエリ登録部１１１の処理の一例を示すフローチャートである。このフローチャートは、前記第一実施形態の図１３に示したフローチャートのステップ１３０３とステップ１３０４の間に、ステップ２００１を加えたもので、その他の処理は図１３と同様である。なお、以下では前記第一実施形態の図１３と同一の処理については重複した説明を省略する。
【０１１２】
第二実施形態のクエリ登録部１１１の処理では、第一実施形態とは異なり、ユーザの入力を送信する登録サーバ１０５から受け取った入力ストリーム用データセットキー２０５と、実行木（コンパイル部で生成）に含まれる各オペレータから、実行可能データ参照テーブル１７０１を生成する（２００１）。生成した実行可能データ参照テーブル１７０１は図１８に示した１８０５、１８０６のようなデータは存在しても構わないし、データが存在しない空のテーブルであってもよい。その他のクエリ登録時の処理については、前記第一実施形態と同様である（１３０１〜１３０７）。
【０１１３】
データ入力部１１０の処理については、前記第一実施形態と同様である。
【０１１４】
図２１Ａ、図２１Ｂは、データ実行部１１２の処理の一例を示すフローチャートである。図２１Ａ、図２１Ｂの処理は、前記第一実施形態の図１５Ａ、図１５Ｂに代わってデータ実行部１１２で行われる処理である。
【０１１５】
データ実行部１１２の処理では、まず実行順序決定部（２１０７）において、入力ストリーム２１４に入力データが存在する限り（２１０１）、１５０２〜１５０５の手順（第一実施形態と同様）に従ってストリームデータ格納キューに入力データを格納し続ける。
【０１１６】
実行順序決定部２１７は、ステップ１５０２〜１５０５については、前記第一実施形態の図１５Ａと同様に、実行順序決定部２１７が、入力ストリーム２１４の入力データ２１３を、入力ストリーム用データセットキー値４０２毎の実行領域２１８Ａ、２１９Ａのストリームデータ格納キュー２２１に格納する。
【０１１７】
図２２Ａ、図２２Ｂは、データ入力部１１０とデータ実行部１１２のタイムチャートを示す。上記データ実行部１１２では、上記ステップ２１０１の処理により、前記第一実施形態とは異なり、データ入力部１１０が出力した入力ストリーム２１４から連続して入力データ７０７、７０６、７０８を取得し、入力データ７０７を実行領域Ａのストリームデータ格納キュー１００２と、入力データ７０６、７０８を実行領域Ｂのストリームデータ格納キュー１０１５にそれぞれ格納する。
【０１１８】
そしてデータ実行部１１２の実行順序決定部２１７は、入力ストリーム２１４に入力データ２１３が存在しなくなったら（図２１Ａの２１０１）、実行木２２８の最初のオペレータを実行オペレータ２２６とする（図２１Ａの２１０２）。
【０１１９】
そしてデータ実行部１１２は、取得した入力データ２１３が属するデータセットの実行領域のストリームデータ格納キュー２２１の先頭データの時刻（複数の入力ストリームのデータを処理するクエリの場合には、複数のストリームデータ格納キュー１００２，１０１５の先頭データの時刻）を比較し、最も時刻が古いデータが存在するストリームデータ格納キューを選択する。そして、データ実行部１１２は、選択したストリームデータ格納キューの入力データ２１３について、実行可能データ参照テーブル１７０１（図１８）の、入力データのデータセット（入力ストリーム用データセットキー４０２）に対応する、実行オペレータの項目に「○」フラグを付加し、実行可能データ参照テーブル１７０１を更新する。
【０１２０】
つまり、実行順序決定部２１７は、ストリームデータ格納キュー１００２，１０１５の入力データ２１３のうち、先頭データの時刻が最も古い時刻の入力データ２１３を抽出し、この入力データ２１３と実行木２２８に対応する実行可能データ参照テーブル１７０１のエントリを、実行可能であることを示す値（例えば、「○」）を書き込んで更新する。
【０１２１】
そして実行順序決定部２１７は、更新後の実行可能データ参照テーブル１７０１を参照し、実行オペレータ２２６において、いずれかのデータセットに実行可能なデータが存在する場合には（図２１Ｂの２１０３）、当該データセットの実行領域上の実行状態２２３を用いて、実行データ２２４を実行オペレータ２２６で処理する（図２１Ｂの１５０９）。
【０１２２】
そして、実行順序決定部２１７は、実行オペレータ２２６が該データセットにおいて実行可能なデータがなくなった場合には、実行可能データ参照テーブル（図１８）の対応する「○」フラグを消すことによりリセットする（図２１Ｂの２１０６）。
【０１２３】
実行順序決定部２１７は、前記実行オペレータ２２６を処理した後、処理結果を、受信サーバ１１７や入力データ受信部２０７において、出力データとして取得する場合には（１５１１）、第一実施形態と同様に単一のストリームにマージする（１５１２）。
実行順序決定部２１７では、以上のステップ２１０３からステップ１５１２の処理を繰り返す。そして実行順序決定部２１７は、ステップ２１０３において実行データ２２４が求められなかった場合には、実行木２２８の次のオペレータを実行オペレータ２２６として（２１０４）、上記ステップ２１０３の処理を行う。そして実行順序決定部２１７は、ステップ２１０４において実行木２２８に次のオペレータが存在しなくなったら、ステップ１５０２の処理に戻って入力ストリーム２１４から入力データ２１３を取得し、上記と同様に処理する。
【０１２４】
例えば、図１９Ａ、図１９Ｂは、入力データ７０７、７０６、７０８を格納したストリームデータ格納キュー２２１（１００２、１０１６）である。データ実行部１１２の実行順序決定部２１７は、ストリームデータ格納キュー２２１へ前記入力データ２１３を格納した後、図２２Ａ、図２２Ｂに示すように、実行木２２８（図６）の最初のオペレータであるオペレータＲＡＮＧＥ６０１を、実行オペレータ２２６として決定する。そして、実行順序決定部２１７が、実行可能データ参照テーブル１７０１（図１８）を参照し、データセット「佐藤ビル」がオペレータＲＡＮＧＥ６０１において実行可能である（フラグ「○」を格納している）ことから、図２２Ａ，図２２Ｂに示すように、実行データを７０６とする。そして実行順序決定部２１７は、実行領域Ｂで実行オペレータ２２６であるＲＡＮＧＥ６０１で実行データ７０６を処理した後、実行可能データ参照テーブル１７０１においてデータセット「佐藤ビル」のオペレータＲＡＮＧＥのフラグ「○」を削除（リセット）し、同データセットのオペレータＧＲＯＵＰ  ＢＹにフラグ「○」を設定し、実行可能データ参照テーブル１７０１を更新する。
【０１２５】
そして実行順序決定部２１７は、図２２Ａ、図２２Ｂで示すように、続いてデータセット「佐藤ビル」のデータ７０８を実行データとし、次にデータセット「田中家」のデータ７０７を実行し、上記と同様に実行可能データ参照テーブル１７０１を更新する。
【０１２６】
そして実行順序決定部２１７では、オペレータＲＡＮＧＥ６０１の実行後、オペレータＲＡＮＧＥ ６０１で実行可能なデータが存在しなくなるため、実行木２２８（図６）の次のオペレータＧＲＯＵＰ  ＢＹ ６０２を実行オペレータ２２６とし、同様に実行データ１６０３，１６０５，１６０１を選択して、オペレータＧＲＯＵＰ  ＢＹ ６０２で処理し、実行可能データ参照テーブル１７０１を上述のように更新する。
【０１２７】
実行順序決定部２１７は、最後にオペレータＩＳＴＲＥＡＭ ６０３で各実行データ１６０４，１６０６，１６０２に対して処理を行った後、第一実施形態と同様に、処理結果を出力データ１２０３、１２０２、１２０４として出力ストリームに格納する。
【０１２８】
なお上記のデータ実行部１１２で行われる処理は、前記非特許文献３に開示されているラウンドロビンのオペレータスケジューリングの中で、各オペレータの実行データ２２４がある限り、同じオペレータを実行オペレータ２２６としていて処理する手法である。上記の他にも一定時間や一定回数まで同一のオペレータを実行オペレータ２２６とするスケジューリング方法や、実行可能なデータが存在するオペレータをランダムに実行オペレータとするスケジューリング方法もあるが、これらのスケジューリング方法についても、本発明では同様に実現可能である。
【０１２９】
なお第二実施形態では実行順序決定部２１７において、ストリームデータ格納キュー２２１を参照する点で非特許文献３と共通しているが、実行領域名参照テーブル２１５及び、実行領域名２１６、実行領域２１８、２１９、実行可能データ参照テーブル１７０１を参照する点で非特許文献３と異なる。また、第二実施形態では非特許文献３とは異なり、ストリームデータ処理サーバ１０８がデータセットキー読込み部２１１、データセットキー２０５、部分的時刻順ソート処理部２０９、データセット変換テーブル２１０、データ出力部２３０を含む点が、構成上の特徴である。
【０１３０】
＜第三実施形態＞
続いて第三実施形態について図面を用いて説明する。第三実施形態は前記第一実施形態、第二実施形態とは異なり、ストリームデータ処理エンジン（本発明ではデータ実行部１１２に相当）を従来例から変更することなしに、同様の効果を実現する。
【０１３１】
図２３Ａ、図２３Ｂは第三実施形態の計算機システムを示すブロック図である。図２３Ａは、ストリームデータ処理サーバ１０８の入出力の関係を示すブロック図である。また、図２３Ｂは、ストリームデータ処理サーバ１０８の詳細なブロック図である。
【０１３２】
第三実施形態では、クエリ登録部１１１において、前記第一実施形態、第二実施形態とは異なり、ユーザ２０４が最大処理データ数２３０１を指定し、指定された最大処理データ数２３０１を、登録サーバ１０５からストリームデータ処理サーバ１０８に送信する。
【０１３３】
そしてクエリ登録部１１１は、ストリーム及びクエリ複製部２３０２において、データセットキー２０５と、ストリーム及びクエリ定義２０６と、最大データセット数２３０１を登録サーバ１０５から受信し、複数ストリーム及びクエリ定義２３０３を生成する。そしてクエリ登録部１１１では、コンパイル部２１２が、複数ストリーム及びクエリ定義２３０３から複数の実行木２２８を生成し、各実行木２２８を複数のデータ実行部１１２に転送する。なお、図２３Ａ、図２３Ｂでは、複数のデータ実行部１１２が３つのデータ実行部＃１〜＃３で構成された例を示すが、任意の数のデータ実行部１１２を備えることができる。
【０１３４】
そしてデータ入力部１１０では、第一実施形態、第二実施形態とは異なり、部分的時刻順ソート処理部２０９において部分的に時刻順でソートした入力データを、データセットとストリームの対応テーブル２３０５に従って、複数のデータ実行部１１２の入力ストリーム２１４にそれぞれ格納する。
【０１３５】
複数のデータ実行部（＃１〜＃３）１１２は、該入力ストリーム２１４を従来のストリームデータ処理システムと同様に処理し、処理結果の出力データ２２９を出力ストリーム２３１に格納して出力する。
【０１３６】
最後にストリームマージ部２３０６において、複数のデータ実行部１１２の出力ストリーム２３１から出力データ２２９を取得して出力キュー２３０７に格納する。
【０１３７】
出力キュー２３０７に格納された出力データ２２９は、受信サーバ１１７や入力データ受信部２０７に送信される。
【０１３８】
図２４は、最大データセット数２３０１の一例を示す。クエリ登録部１１１が最大データセット数２３０１は、入力ストリーム用データセットキー４０２に設定可能な値の数を表す。図中２３０１では、最大データセット数は３であり、入力ストリーム用データセットキーが「設置場所」であることから、設置場所の数が最大で３であることを示す。後述するように最大データセット数２３０１は、データ実行部１１２の入力ストリーム２１４の数を示す。
【０１３９】
図２５は、データ入力部１１０のデータセットとストリームの対応テーブル２３０５の一例を示す。第三実施形態では、データ入力部１１０が、入力ストリーム用データセットキーの値ごとに、複数のデータ実行部＃１〜＃３に対応する異なる入力ストリーム２１４を生成して入力データを格納する。データセットとストリームの対応テーブル２３０５では、入力ストリーム用データセットキー値２５０１と、入力ストリーム用データセットキー値を有する入力データを格納する入力ストリーム２５０２の対応関係を示す。例えば図中のエントリ２５０３は、入力ストリーム用データセットキー値２５０１が「田中家」の入力データは、入力ストリーム「電気メータ１」に格納することを示す。またエントリ２５０４では、入力ストリーム用データセットキー値２５０１が「佐藤ビル」の入力データは、入力ストリーム「電気メータ２」に格納することを示す。またエントリ２５０５に、入力ストリーム「電気メータ３」に格納する入力データが存在しないことを示す。図示の例では、入力ストリーム「電気メータ１」はデータ実行部＃１の入力ストリーム２１４に対応し、入力ストリーム「電気メータ２」は、データ実行部＃２の入力ストリーム２１４に対応し、入力ストリーム「電気メータ３」は、データ実行部＃３の入力ストリーム２１４に対応する。
【０１４０】
図２６は、出力キュー２３０７の一例を示す。出力キュー２３０７は、受信サーバ１１７や入力データ受信部２０７に送信する出力データ２２９を格納するキューである。図２６では、出力キュー２３０７に、出力データ１２０２〜１２０４が格納されている例を示す。
【０１４１】
図２７Ａ〜図２７Ｆは、クエリ登録部１１１の複製ストリーム及びクエリ定義２３０３の一例を示す。複製ストリーム及びクエリ定義２３０３は、ユーザが登録サーバ１０５で指定したストリーム及びクエリ定義２０６（図３Ａ、図３Ｂ）のストリーム定義と、クエリ定義をストリーム及びクエリ複製部２３０２で複製し、複製後のストリーム名と、クエリ名を変更したストリーム定義と、クエリ定義である。
【０１４２】
図２７Ａ、図２７Ｃ，図２７Ｅは、図２５に示したデータセットとストリームの対応テーブル２３０５の電気メータ１〜３ストリーム２７０１、２７０３、２７０５を示し、各ストリームの定義は、ストリーム及びクエリ複製部２３０２が、図３Ａの電気メータストリーム定義３０１を３つに複製し、それぞれストリーム名を変更したストリーム定義である。
【０１４３】
また、図２７Ｂ、図２７Ｄ，図２７Ｆは、電気メータ１〜３ストリームに対応する使用電力集計１〜３クエリ２７０２、２７０４、２７０６を示し、ストリーム及びクエリ複製部２３０２が、図３Ｂに示した使用電力集計クエリ定義３０２を３つに複製し、クエリ名を変更したクエリ定義である。
【０１４４】
図２８は、クエリ登録部１１１の処理の一例を示すフローチャートである。クエリ登録部１１１は、データセットキー読込み部２１１で登録サーバ１０５からデータセットキー２０５を受信する（２８０１、１３０２）。次に、クエリ登録部１１１は、第一実施形態とは異なり、図３Ｃに示したデータセットキー２０５の入力ストリーム用データセットキー３０４からデータセット＆ストリーム対応テーブル２３０５を生成する（２８０２）。生成されたデータセット＆ストリーム対応テーブル２３０５は、図２５のエントリ２５０３〜２５０５のようなデータは存在しても構わないし、データが存在しない空のテーブルであっても構わない。
【０１４５】
そして、クエリ登録部１１１は前記第一実施形態の図１３と同様にステップ１３０４，１３０５でデータセット変換テーブル２１０を生成する。クエリ登録部１１１は、データセットキー読込み部２１１の処理が終了した後（２８０３）、ストリーム及びクエリ複製部２３０２で、登録サーバ１０５からストリーム及びクエリ定義２０６と最大データセット数２３０１を受信し、定義されたストリームと、クエリを最大データセット数２３０１の値に応じた数だけ複製し、名称を変更して複数ストリーム及びクエリ定義２３０３を生成する（２８０４）。
【０１４６】
そしてクエリ登録部１１１のコンパイル部２１２は、複製された複数のストリーム及びクエリ定義２３０３をコンパイルして、複数の実行木２２８を生成する（２８０５）。そして、生成された複数の実行木２２８を異なるデータ実行部＃１〜＃３におけるオペレータ処理部２２７に格納する。例えば、前記第一実施形態の図３Ａ、図３Ｂと同様に、ストリーム定義３０１、クエリ定義３０２が設定され、最大データセット数２３０１は図２４で示した「３」の場合には、クエリ登録部１１１は、図２７Ａ、図２７Ｃ、図２７Ｅに示した複製ストリーム（電気メータ１〜３ストリーム）２７０１，２７０３，２７０５と、図２７Ｂ、図２７Ｄ、図２７Ｆのクエリ定義（使用電力集計１〜３クエリ）２７０２，２７０４，２７０６を生成する。そしてコンパイル部２１２において、複製ストリーム及びクエリ定義２７０１〜２７０６から、実行木２２８（図６）を３つ生成する。生成された各実行木２２８はクエリ登録部１１１から各データ実行部＃１〜＃３に出力される。
【０１４７】
図２９は、データ入力部１１０の処理の一例を示すフローチャートである。データ入力部１１０では、前記第一実施形態の図１４のステップ１４０２〜１４０９と同様に、データセット変換テーブル２１０を用いて入力データの受信処理を行い、入力ストリーム用データセットキー値４０２（図４）毎に入力データを時刻順にソートする（１４１１）。そして、データ入力部１１０の部分的時刻順ソート処理部２０９で部分的時刻順ソート処理が終了した後（２９０２）、ストリーム振分け処理部２３０４の処理を開始する（２９０３）。
【０１４８】
ストリーム振分け処理部２３０４では、時刻が最古の入力データの入力ストリーム用データセットキー値２５０１から、データセット＆ストリーム対応テーブル２３０５を参照し、該データが属する入力ストリーム２５０２を抽出し（２９０５）、該データを入力ストリーム２１４に格納する（２９０７）。ステップ２９０５において、ストリーム振分け処理部２３０４は、該データが属する入力ストリーム２５０２が抽出されない場合には、データセット＆ストリーム対応テーブル２３０５に、該データの入力ストリーム用データセットキー値２５０１と、データセットが割当てられていない入力ストリーム名２５０２を追加し（２９０６）、当該入力ストリームに該データを格納する（２９０７）。該データは従来のストリームデータ処理システムと同様に入力ストリーム用データセットキー値２５０１毎の入力ストリーム２５０２を複数のデータ実行部＃１〜＃３で処理し、処理結果を出力ストリーム２３１にそれぞれ格納する。
【０１４９】
図３１は、データ入力部１１０とデータ実行部（＃１，＃２）１１２、ストリームマージ部２３０６のタイムチャートを示す。データ入力部１１０において、入力データ７０６〜７０８を受信し、データセット「田中家」の入力データ７０７と、データセット「佐藤ビル」の入力データ７０６、７０８を、データセット＆ストリーム対応テーブル２３０５（図２５）に従って、独立したデータ実行部＃１、＃２の入力ストリーム２５０２（２１４）に格納する。そしてそれぞれのデータ実行部＃１、＃２では、出力ストリーム２３１に、出力データ１２０４及び、出力データ１２０３、１２０２を格納する。
【０１５０】
図３０は、ストリームマージ部２３０６の処理の一例を示すフローチャートである。ストリームマージ部２３０６（３００１）では、データ実行部＃１〜＃３のいずれかの出力ストリーム２３１に出力データが存在する限り（３００２）、該出力データを出力ストリームに関係なく単一の出力キュー２３０７に格納し続ける（３００３）。例えば、図２６に示すように、出力キュー２３０７に、入力データ７０６〜７０８の処理結果を、出力データ１２０２〜１２０４として格納する。なお、出力キュー２３０７に格納された出力データ１２０２〜１２０４は、所定のタイミング（例えば、所定の周期）で受信サーバ１１７に送信される。
【０１５１】
以上で記述した第三実施形態は、従来のストリームデータ処理エンジン（データ実行部＃１〜＃３）を変更することなしに、前記第一実施形態と同様の効果を実現することができる。上記の他に、最大データセット数２３０１を指定せず、データ実行部＃１〜＃３での処理の実行時に入力ストリーム用データセットキー値２５０１の数の増加に従って、動的にストリーム及びクエリの定義を複製する。そして複製したストリーム及びクエリをコンパイルし、生成した実行木２２８を登録することも可能である。
【０１５２】
なお、上記第一〜第三の実施形態において、データセットキー２０５やストリーム及びクエリ定義２０６等の設定情報を登録サーバ１０５から受信する例を示したが、ストリームデータ処理サーバ１０８に入力装置を設け、当該入力装置から設定情報を受け付けてもよい。
【０１５３】
以上、本発明を添付の図面を参照して詳細に説明したが、本発明はこのような具体的構成に限定されるものではなく、添付した請求の範囲の趣旨内における様々な変更及び同等の構成を含むものである。
【産業上の利用可能性】
【０１５４】
本発明は、時刻を含む入力データに対してストリームデータ処理を行う計算機システムに適用することができる。
【図１】
2012046316.tif 000003
【図２Ａ】
2012046316.tif 000004
【図２Ｂ】
2012046316.tif 000005
【図３Ａ】
2012046316.tif 000006
【図３Ｂ】
2012046316.tif 000007
【図３Ｃ】
2012046316.tif 000008
【図４】
2012046316.tif 000009
【図５】
2012046316.tif 000010
【図６】
2012046316.tif 000011
【図７】
2012046316.tif 000012
【図８】
2012046316.tif 000013
【図９】
2012046316.tif 000014
【図１０Ａ】
2012046316.tif 000015
【図１０Ｂ】
2012046316.tif 000016
【図１１】
2012046316.tif 000017
【図１２】
2012046316.tif 000018
【図１３】
2012046316.tif 000019
【図１４Ａ】
2012046316.tif 000020
【図１４Ｂ】
2012046316.tif 000021
【図１５Ａ】
2012046316.tif 000022
【図１５Ｂ】
2012046316.tif 000023
【図１６Ａ】
2012046316.tif 000024
【図１６Ｂ】
2012046316.tif 000025
【図１７Ａ】
2012046316.tif 000026
【図１７Ｂ】
2012046316.tif 000027
【図１８】
2012046316.tif 000028
【図１９Ａ】
2012046316.tif 000029
【図１９Ｂ】
2012046316.tif 000030
【図２０】
2012046316.tif 000031
【図２１Ａ】
2012046316.tif 000032
【図２１Ｂ】
2012046316.tif 000033
【図２２Ａ】
2012046316.tif 000034
【図２２Ｂ】
2012046316.tif 000035
【図２３Ａ】
2012046316.tif 000036
【図２３Ｂ】
2012046316.tif 000037
【図２４】
2012046316.tif 000038
【図２５】
2012046316.tif 000039
【図２６】
2012046316.tif 000040
【図２７Ａ】
2012046316.tif 000041
【図２７Ｂ】
2012046316.tif 000042
【図２７Ｃ】
2012046316.tif 000043
【図２７Ｄ】
2012046316.tif 000044
【図２７Ｅ】
2012046316.tif 000045
【図２７Ｆ】
2012046316.tif 000046
【図２８】
2012046316.tif 000047
【図２９】
2012046316.tif 000048
【図３０】
2012046316.tif 000049
【図３１Ａ】
2012046316.tif 000050
【図３１Ｂ】
2012046316.tif 000051
【国際調査報告】
2012046316.tif 000052
2012046316.tif 000053
2012046316.tif 000054
2012046316.tif 000055