(57)【要約】【課題】  輻輳状態が発生した時に、自律的に全てのデータ処理プログラムが輻輳したバッファ装置に集中してしまうのを防ぐ。【解決手段】  本発明は、バッファ装置で輻輳が発生すると、輻輳が発生しているバッファ装置を優先的に選択するルールで動作するデータ処理プログラムだけでなく、輻輳が発生してもそれ以前と同じ順序でバッファ装置の選択を行うルールで動作するデータ処理プログラムを混在させることで、データ処理プログラムが輻輳したバッファ装置への集中を防ぐ。【選択図】  図１
2014225111.tif 000002
【特許請求の範囲】
【請求項１】
ストリーム処理システムにおいて、データソースからの入力データを一時的に保持する複数のバッファ装置に蓄積されたデータを自律的に処理する複数のデータ処理プログラムを有するデータ処理装置であって、
前記バッファ装置に輻輳が発生すると、輻輳が発生しているバッファ装置を優先的に選択するルールで動作するデータ処理プログラムと、輻輳が発生しても、該輻輳が発生する前と同じ順序でバッファ装置の選択を行うルールで動作するデータ処理プログラムの両方を制御する手段を有することを特徴とするデータ処理装置。
【請求項２】
前記バッファ装置から輻輳通知を取得すると、当該輻輳バッファ装置の状態情報を保持する状態管理手段と、
前記ルールとして、輻輳が発生していれば輻輳が発生しているバッファ装置からデータ取得先のバッファ装置を決定するルールAと、輻輳が発生していなければ全バッファ装置からデータ取得先のバッファ装置を決定するルールBを含む第１のルールと、該バッファ装置の輻輳には非対応の第２のルールを設定し、要求に応じてルールを通知するルール構成設定手段と、
前記データ処理プログラムからバッファ装置選択要求が入力されると、前記状態管理手段の前記輻輳バッファ装置の状態情報を参照して、前記ルール構成設定手段から取得した前記第１のルールと前記第２のルールのうち、指定されたルールに応じて選択したバッファ装置を前記データ処理プログラムに通知するバッファ装置選択手段と、
を有し、
前記データ処理プログラムは、
前記バッファ装置選択手段で選択されたバッファ装置からデータを取得する
請求項１記載のデータ処理装置。
【請求項３】
前記バッファ装置から輻輳通知を取得すると、当該輻輳バッファ装置の状態情報を保持する状態管理手段と、
前記バッファ装置に輻輳が発生している場合に、輻輳に対応すべきか否かを表すα（０≦α≦１）値の割合を設定する輻輳対応設定手段と、
前記状態管理手段の前記状態情報を参照して、輻輳が発生していない場合は、複数のバッファ装置を、どのような順序で前記バッファ装置のデータを取得するかを定めたルールに応じて巡回してデータを取得し、輻輳が発生している場合は、前記輻輳対応設定手段から前記α値を取得して乱数判定により、バッファ装置を選択し、選択したバッファ装置を前記データ処理プログラムに通知するバッファ装置選択手段と、
を有し、
前記バッファ装置選択手段は、
０〜１の間の乱数を発生し、当該乱数が０以上α未満の場合は輻輳状態のバッファ装置をデータ取得先のバッファ装置とし、当該乱数がα以上１以下の場合は全バッファ装置を選択する手段を含む
請求項１記載のデータ処理装置。
【請求項４】
前記バッファ選択手段は、
輻輳継続時間または輻輳しているバッファ装置の数に基づいて前記α値を大きくし、段階的に前記データ処理プログラムの数を増やす手段
を含む請求項３記載のデータ処理装置。
【請求項５】
ストリーム処理システムにおいて、データソースからの入力データを一時的に保持する複数のバッファ装置に蓄積されたデータを自律的に処理する複数のデータ処理プログラムを有するデータ処理方法であって、
データ処理装置が、
前記バッファ装置に輻輳が発生すると、輻輳が発生しているバッファ装置を優先的に選択するルールで動作するデータ処理プログラムと、輻輳が発生しても、該輻輳が発生する前と同じ順序でバッファ装置の選択を行うルールで動作するデータ処理プログラムの両方を制御することを特徴とするデータ処理方法。
【請求項６】
前記データ処理装置が、状態管理手段、ルール構成設定手段、バッファ装置選択手段、前記データ処理プログラムを有し、
前記状態管理手段が、前記バッファ装置から輻輳通知を取得すると、当該輻輳バッファ装置の状態情報を保持する状態管理ステップと、
前記ルール構成設定手段が、前記ルールとして、輻輳が発生していれば輻輳が発生しているバッファ装置からデータ取得先のバッファ装置を決定するルールAと、輻輳が発生していなければ全バッファ装置からデータ取得先のバッファ装置を決定するルールBを含む第１のルールと、該バッファ装置の輻輳には非対応の第２のルールを設定し、要求に応じてルールを通知するルール構成設定ステップと、
前記バッファ装置選択手段が、前記データ処理プログラムからバッファ装置選択要求が入力されると、前記輻輳バッファ装置の状態情報を参照して、前記ルール構成設定手段から取得した前記第１のルールと前記第２のルールのうち、指定されたルールに応じて選択したバッファ装置を前記データ処理プログラムに通知するバッファ装置選択ステップと、
を行う請求項５記載のデータ処理方法。
【請求項７】
前記データ処理装置が、状態管理手段、輻輳対応設定手段、バッファ装置選択手段、前記データ処理プログラムを有し、
前記状管理手段が、前記バッファ装置から輻輳通知を取得すると、当該輻輳バッファ装置の状態情報を保持する状態管理ステップと、
前記輻輳対応設定手段が、前記バッファ装置に輻輳が発生している場合に、輻輳に対応すべきか否かを表すα（０≦α≦１）値の割合を設定する輻輳対応設定ステップと、
前記バッファ装置選択手段が、前記状態情報を参照して、輻輳が発生していない場合は、複数のバッファ装置を、どのような順序で前記バッファ装置のデータを取得するかを定めたルールに応じて巡回してデータを取得し、輻輳が発生している場合は、前記輻輳対応設定手段から前記α値を取得して乱数判定により、バッファ装置を選択し、選択したバッファ装置を前記データ処理プログラムに通知するバッファ装置選択ステップと、
を行い、
前記バッファ装置選択ステップにおいて、
０〜１の間の乱数を発生し、当該乱数が０以上α未満の場合は輻輳状態のバッファ装置をデータ取得先のバッファ装置とし、当該乱数がα以上１以下の場合は全バッファ装置を選択する
請求項５記載のデータ処理方法。
【請求項８】
コンピュータを、
請求項１乃至４のいずれか１項に記載のデータ処理装置の各手段として機能させるためのデータ処理プログラム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、データ処理装置及び方法及びプログラムに係り、特に、ストリーム処理システムにおけるバッファ装置に蓄積されたデータを自律的に処理するデータ処理を最適化するためのデータ処理装置及び方法及びプログラムに関する。
【背景技術】
【０００２】
複数のバッファ装置に蓄積されたストリームデータを、複数のデータ処理プログラムが並列に処理するシステムが存在する。
【０００３】
それぞれのデータ処理プログラムは、並列にデータ処理を行うために、どのバッファ装置からデータを取得するかを決定する必要がある。データ処理プログラムが、どのバッファ装置を選択してデータを取得、処理するかを決定するルール（以下、「PULL戦略」と記す）には、全てのバッファ装置を順番に選択を行う方法や、ランダムにバッファ装置を選択する方法が考えられる。
【０００４】
バッファ装置に蓄積されるデータ量に対して、それを処理するデータ処理プログラムの処理能力が十分な場合には、これら単純なPULL戦略も有効に動作し、並列処理を問題なく実行できる。
【０００５】
また、一部のバッファ装置へのデータ入力が多く、輻輳した状態が発生している場合には、自律分散されたデータ処理装置は、バッファ装置の破綻を防ぐために輻輳したバッファ装置に蓄積されたデータを集中的に処理して輻輳を解消する必要がある。データ処理装置はシステム内の状態を監視して、輻輳が発生している場合には、該当のバッファ装置を集中的に処理するよう内部状態を切り替えることでこれを実現している（例えば、特許文献1参照）。
【０００６】
図１は、従来技術におけるデータ処理プログラムの動作のフローチャートである。
【０００７】
データ処理プログラムは、輻輳が発生しているか否かを判定し（ステップ１０）、輻輳が発生していない場合（図２）には、規定の手順により取得先バッファ装置を決定し（ステップ１１）、バッファ装置からデータを取得して（ステップ１３）、データ処理を行う（ステップ１４）。輻輳が発生している場合（図３）には、輻輳したバッファ装置を取得先バッファ装置に決定し（ステップ１２）、当該バッファ装置からデータを取得し（ステップ１３）、データ処理を行う（ステップ１４）。
【０００８】
図３に示すように、バッファ装置Ａで輻輳が発生している場合、バッファ装置Ａは、データ処理装置に輻輳を通知し、データ処理装置では輻輳をできるだけ早く解消するためにバッファ装置Aのデータを優先的に取得、処理しようとする。このときPULL戦略次第では、バッファ装置Aの輻輳が解消するまでバッファ装置Ｂの処理は行われない場合がある。
【０００９】
このようなPULL戦略としてはさまざまなものが考えられるが、その一例を以下に示す。
【００１０】
・PULL_RR：ラウンドロビンのPULL戦略であり、当該戦略を指定した場合は、複数あるバッファ装置から順にデータ取得を試みる。
【００１１】
・PULL_RAND：当該戦略を指定した場合は、複数あるバッファ装置を選択する際に、疑似乱数Ｒを生成し、疑似乱数Ｒをバッファ装置の数ｎで割った余り（R%n等）に基づいてどのバッファ装置からデータ取得を試みるかを決定する。
【００１２】
・PULL_RR_JAM：輻輳が発生していない場合はPULL_RRと同様の動作をするが、輻輳が発生した場合は輻輳しているバッファ装置から優先してデータを試みるPULL戦略である。
【００１３】
・PULL_BIAS_RR：各バッファ装置を指定した割合で取得するラウンドロビンのPULL戦略であり、別途、どのバッファ装置をどの割合で取得するかを設定する。
【００１４】
PULL戦略の選択は、ストリーム処理のさまざまな要件に応じて選択する必要がある。要件の例として、データ処理完了までのレイテンシがあげられる。レイテンシをできるだけ低く抑えたいようなケースでは、データ処理プログラムが不足することのないように十分な数を確保することが最も重要である。
【００１５】
しかし、データ処理プログラム自体は不足していなくても、バッファ装置の数が大量に存在する時は、データ処理プログラムがバッファ装置を選択（巡回）するタイミングが遅れレイテンシが増大するなどの影響を与える。
【００１６】
バッファ装置に追記されると確実に一定期間内に処理をする必要がある場合に、PULL_RANDのようなPULL戦略を使用した場合、特定のバッファ装置が確率的にいつまでも選択されない危険性がある。また輻輳に対応したPULL_RR_JAMのようなPULL戦略を選択した場合には、別のバッファ装置が輻輳している場合に当該バッファ装置が選択されない危険性がある。
【００１７】
このような場合には、確実にすべてのバッファ装置を巡回するPULL_RRのようなPULL戦略のほうが適している。
【００１８】
また別のケースとして、普段は入力が少ないが、定期的、不定期などで一時的に大量のデータが到達するようなバッファ装置が存在して、このバッファ装置は優先的に処理をする必要があるといった要件がある。
【００１９】
この場合、各バッファ装置に対して、一定の割合でデータ取得を試みるPULL_RANDやPULL_RRのようなPULL戦略では、大量データが到達したバッファ装置のデータを処理しきるまでに時間がかかってしまう危険性がある。
【００２０】
このような場合には、PULL_RR_JAMのように輻輳に対応したPULL戦略を使用すると、大量にデータが追記されたバッファ装置を効率的に選択し、輻輳の解消までの時間を短くすることが可能である。
【００２１】
逆のケースとして、常に入力されているバッファ装置のデータの処理は性能を維持する必要があるが、普段は入力が少なくて、定期的、不定期などで一時的に大量のデータが到達するようなバッファ装置が存在して、このバッファ装置はベストエフォートで処理を行うというケースも存在する。
【００２２】
このような場合は、PULL_RR_JAMのように輻輳に対応したPULL戦略のみを使用すると、データ処理プログラムが輻輳したバッファ装置の処理への対応に占有され、常に入力されているバッファ装置が選択されず、データ処理が遅れてしまう危険性がある。
このような場合には、すべてのバッファ装置を選択対象とするPULL_RRやPULL_RANDのようなPULL戦略のほうが適している。
【００２３】
上述のように、一つのPULL戦略のみでデータ処理を実施しようとした場合には、状況によって要件が満たせなくなることがわかる。また、複数の入力特性や処理までのレイテンシの要件が混在するシステムでは、常に要件を満たせないことがわかる。
【先行技術文献】
【特許文献】
【００２４】
【特許文献１】特許第4764300号公報
【発明の概要】
【発明が解決しようとする課題】
【００２５】
しかしながら、上記の従来技術においては、各データ処理プログラムが自律分散しているため、各データ処理プログラムが輻輳したバッファ装置に集中する。その結果、輻輳していないバッファ装置の処理を実施するデータ処理プログラムが大幅に減少して処理が遅れるという最適化が過剰な状態が発生する。
【００２６】
図４に沿って従来の課題を説明する。
【００２７】
データ入力の集中（輻輳）が発生すると、バッファ装置Ａは、バッファ溢れを防ぐために、データ入力の集中をデータ処理装置に対して通知する（図４（ａ））。
【００２８】
通知を受けたデータ処理装置のデータ処理プログラムがデータ入力の集中したバッファ装置から優先的にデータを取得してデータ処理を実行する（図４（ｂ））。
【００２９】
上記のような輻輳時の対応は、輻輳を短時間で解消するために必要な機能であるが、過度にデータ処理プログラムからのアクセスが集中してしまい、データ入力の集中したバッファ装置以外のバッファ装置のデータが処理できないためレイテンシが上昇する。
【００３０】
本発明は、上記の点に鑑みなされたもので、複数のバッファ装置に蓄積されたストリームデータを複数の自律分散したデータ処理プログラムによって並列に処理するシステムにおいて、輻輳状態が発生した時に、自律的に全てのデータ処理プログラムが輻輳したバッファ装置に集中してしまうのを防ぐことが可能なデータ処理装置及び方法及びプログラムを提供することを目的とする。
【課題を解決するための手段】
【００３１】
上記の課題を解決するため、本発明の一態様によれば、ストリーム処理システムにおいて、データソースからの入力データを一時的に保持する複数のバッファ装置に蓄積されたデータを自律的に処理する複数のデータ処理プログラムを有するデータ処理装置であって、
前記バッファ装置に輻輳が発生すると、輻輳が発生しているバッファ装置を優先的に選択するルールで動作するデータ処理プログラムと、輻輳が発生しても、該輻輳が発生する前と同じ順序でバッファ装置の選択を行うルールで動作するデータ処理プログラムの両方を制御する手段を有するデータ処理装置が提供される。
【発明の効果】
【００３２】
上記のように、本発明は、自律分散したデータ処理プログラムに対して、状態管理情報を保持し、かつ、予めバッファ装置の輻輳状態に対応する条件を設定しておく。この条件の設定によって、バッファ装置における輻輳状態に対して、輻輳したバッファ装置に集中するデータ処理プログラムだけでなく、輻輳状態が発生してもバッファ装置に集中しないデータ処理プログラムを混在させ、輻輳していないバッファ装置の処理が行われなくなるのを防ぐことで、輻輳状態が発生した時に、自律的に全てのデータ処理プログラムが輻輳したバッファ装置に集中してしまうのを防ぐことが可能となる。
【図面の簡単な説明】
【００３３】
【図１】従来技術のデータ処理プログラムの動作のフローチャートである。
【図２】従来技術における輻輳が発生していない状態を示す図である。
【図３】従来技術における輻輳が発生した状態を示す図である。
【図４】従来技術の課題を説明するための図である。
【図５】本発明の前提とするシステム構成例である。
【図６】本発明の第１の実施の形態におけるシステム構成図である。
【図７】本発明の第１の実施の形態におけるバッファ装置におけるデータ受信とデータ処理装置への輻輳状態の通知処理のフローチャートである。
【図８】本発明の第１の実施の形態におけるデータ処理でのデータ処理全体のフローチャートである。
【図９】本発明の第１の実施の形態におけるデータ処理装置でのデータ処理のフローチャートである。
【図１０】本発明の第１の実施の形態におけるデータ処理装置でのバッファ装置選択のフローチャートである。
【図１１】本発明の第１の実施の形態におけるデータ処理装置及びバッファ装置でのデータ取得処理のフローチャートである。
【図１２】本発明の第２の実施の形態におけるシステム構成図である。
【図１３】本発明の第２の実施の形態におけるバッファ装置におけるデータ受信とデータ処理装置への輻輳状態の通知処理のフローチャートである。
【図１４】本発明の第２の実施の形態におけるデータ処理装置でのデータ処理全体のデータ処理のフローチャートである。
【図１５】本発明の第２の実施の形態におけるデータ処理装置でのデータ処理のフローチャートである。
【図１６】本発明の第２の実施の形態におけるデータ処理装置でのバッファ装置選択のフローチャートである。
【図１７】本発明の第２の実施の形態におけるデータ処理装置でのデータ取得処理のフローチャートである。
【発明を実施するための形態】
【００３４】
以下、図面と共に本発明の実施の形態を説明する。
【００３５】
本発明の実施の形態を説明する前に、本発明の前提とするシステムについて説明する。
【００３６】
図５は、本発明の前提とするシステム構成例である。
【００３７】
本発明は、PULL型のストリーム分散処理システムであり、データソース２０からの入力データを一時的に保持する部分（バッファ装置１０）とデータ処理装置１００が分離されていてかつ、バッファ装置１０、データ処理装置１００が複数のハードウェア上に存在する。
【００３８】
また、一つのデータ処理装置１００のハードウェア上には、複数のデータ処理プログラムがプロセスまたはスレッドの形態で存在する（以下、プロセスまたはスレッドの形態で存在するデータ処理装置のプログラムを「データ処理プログラム」と呼ぶ）。当該データ処理プログラムは、バッファ装置を一定の順序（PULL戦略）で自律的にポーリングして、データが保持されていれば、これを取得して処理を行う。
【００３９】
データソース２０からのデータ入力レートが一定とは限らないため、データ入力が輻輳した場合は、輻輳している旨をシステム内で共有することで、データ処理プログラムは、輻輳したバッファ装置のデータ処理を優先して行う。
【００４０】
本発明は、上記の処理を行うことにより、PULL型のストリーム処理システムにおいて、複数のデータソースが処理されるまでのレイテンシを悪化させることなく処理する。
【００４１】
[第1の実施の形態]
本実施の形態では、データ処理装置内のデータ処理プログラムが、選択したPULL戦略に基づいて処理の度にバッファ装置の選択を行う。本実施の形態では、PULL戦略は、データ処理装置内のデータ処理プログラムの起動時に決定される。
【００４２】
図６は、本発明の第1の実施の形態におけるシステム構成を示す。
【００４３】
同図に示すシステムは、バッファ装置１０A、１０B、データソース２０、データ処理装置１００１、１００２を有する。
【００４４】
各バッファ装置１０は、蓄積部１１、閾値情報１２、通信部１３を具備し、各データ処理装置１００は、通信部１１０、PULL戦略構成設定部１２０、バッファ装置選択部１３０、状態管理部１４０、データ処理プログラム１５０を有する。バッファ装置選択部１３０は、内部に保持するPULL戦略に基づいてバッファ装置１０を選択する。
【００４５】
状態管理部１４０には、バッファ装置のリスト（バッファ装置識別情報）と輻輳バッファ装置のリスト（輻輳バッファ装置識別情報）を有する。
【００４６】
また、バッファ装置１０の前段には、データソース２０で発生したデータをバッファ装置１０に追記するプログラム（以下、追記クライアント２１）が存在することを前提とする。
【００４７】
各データ処理プログラム１５０は、選択したPULL戦略に基づいて、処理の度にデータを取得（フェッチ）するバッファ装置１０を選択して、データを取得する。本実施の形態におけるPULL戦略は、輻輳が発生したら輻輳したバッファ装置を取得先に設定するものと、輻輳が発生しても全バッファ装置を対象とするものの両方を用いる。
【００４８】
以下、図７〜図１１のフローチャートに沿って、本実施の形態の動作を説明する。
【００４９】
まず、バッファ装置１０は、外部の追記プログラム等から各データDが追記され、輻輳している場合には、データ処理装置１００に通知する。この動作の流れを図７のフローチャートに示す。
【００５０】
図７は、本発明の第１の実施の形態におけるバッファ装置におけるデータ受信とデータ処理装置への輻輳状態の通知処理のフローチャートである。
【００５１】
図７のフローチャートの開始は、前述の追記クライアントが通信部１３へデータ送信した時点からを記述してある。データDは、データソース２０で発生したストリーム処理の対象データであり、ある時刻tのデータDをDtと表す。
【００５２】
バッファ装置１０は、Dtを通信部１３で受信する（ステップ１０１）。通信部１３は、任意の通信プロトコルで受信可能であるが、通常はIPプロトコルが使用される。通信部１３が受信したDtは、一旦このバッファ装置１０の蓄積部１１にデータ保存されて（ステップ１０２）、データ処理装置１００によって処理されるのを待つ。当該データDtが到着する以前のデータDt-1や以後のデータDt+1と同様に受信したデータは、バッファ装置１０の蓄積部１１に保存される。このとき、蓄積部１１は、閾値情報１２を参照し（ステップ１０３）、その時点で蓄積部に蓄積されているデータの量が閾値情報を超えている場合は、輻輳通知としてデータ処理装置へ通知を行う（ステップ１０４，１０５）。
【００５３】
閾値情報の指定方法としては、追記されたデータDの数やデータDの容量の合計値などが利用可能である。
【００５４】
ステップ１０５の輻輳通知は、任意のプロトコルが使用可能であるが、通常はIPプロトコルが使用される。さらに、輻輳通知は、バッファ装置１０のどの蓄積部１１が閾値を超えているかを一意に特定するための情報を含んでいる必要がある。
【００５５】
例として、IPプロトコルを利用した実装においては、輻輳通知には、輻輳しているバッファ装置のIPアドレス、通信ポート番号を含んでいるため、データ処理装置１００は、どのバッファ装置１０が輻輳しているかを知ることができる。
【００５６】
また、別の例として、本システム内で、各バッファ装置１０を識別可能なバッファ装置ID情報とバッファ装置が動作しているIPアドレス、通信ポート番号との対応表を管理していて、バッファ装置ID情報のみを含め、各データ処理装置１００側でバッファ装置ID情報からIPアドレスと通信ポートを特定することでも、データ処理装置１００は、どのバッファ装置が輻輳しているかを知ることができる。
【００５７】
データ処理装置１００は、輻輳通知を受信すると（ステップ１０６）、一旦輻輳しているバッファ装置１０の情報を状態管理部１４０に保存する（ステップ１０７）。
【００５８】
データ処理装置１００は、状態管理部１４０およびPULL戦略構成設定部１２０の内容に応じてバッファ装置１０からデータを取得してデータ処理を行う。この時のデータ処理装置におけるデータ処理順序制御を図８のフローチャートに示す。
【００５９】
図８は、本発明の第１の実施の形態におけるデータ処理装置でのデータ処理全体のフローチャートである。
【００６０】
図８のフローチャートの開始は、データ処理プログラム１５０の開始した時点から記述してある。データ処理プログラムの開始時にはPULL戦略構成の取得を行う。
【００６１】
データ処理プログラム１５０が起動されると（ステップ２０１）、PULL戦略構成設定部１２０では、バッファ装置１０の数および状態管理部１４０の状態に応じて、当該データ処理プログラム１５０がどのような順序でバッファ装置１０のデータを取得するか（PULL戦略）を記述し、メモリ（図示せず）に保持する（ステップ２０２）。
【００６２】
以下にPULL戦略の例を示す。
【００６３】
（１）  PULL戦略の例として、ラウンドロビンのPULL戦略（以下、PULL_RR）がある。PULL_RRを指定した場合は、複数あるバッファ装置から順にデータを取得する。バッファ装置１０の識別子をBUFF_1、BUFF_2 ...BUFF_nとした場合、データ処理プログラムは、BUFF_1、BUFF_2 ...BUFF_nの順にデータ取得を試みる。
【００６４】
（２）ランダムのPULL戦略（以下、PULL_RAND）がある。PULL_RANDを指定した場合は、複数あるバッファ装置１０を選択する際に、データ処理プログラム１５０は、疑似乱数Rを生成し、疑似乱数Rをバッファ装置の数nで割った余り（R % n等）でどのバッファ装置からデータ取得を試みるかを決定する。バッファ装置のバッファ装置の識別子をBUFF_1、BUFF_2 ... BUFF_nとしてa = R % nであった場合は、BUFF_aが選択されデータ取得を試みる。
【００６５】
（３）各バッファ装置１０を指定した割合で取得するラウンドロビンのPULL戦略（以下、PULL_BIAS_RR）がある。PULL_BIAS_RRを指定した場合は、別途、どのバッファ装置１０をどの割合で取得するかを設定する必要がある。例として、バッファ装置BUFF_1とバッファ装置BUFF_2をそれぞれ１：２で取得するよう設定した場合は、データ処理プログラム１５０が3回バッファ装置を選択してデータ取得を試みる間に、1回はBUFF_1を選択してデータ取得を試み、また、2回はBUFF_2を選択してデータ取得を試みる。
【００６６】
（４）状態管理部１４０に輻輳しているバッファ装置１０を表す情報が存在しない時はラウンドロビンのPULL戦略（PULL_RR）と同じ動作をするが、状態管理部１４０に輻輳しているバッファ装置が存在していることを表す情報が保存されている場合には、輻輳しているバッファ装置から優先してデータ取得を試みるPULL戦略（以下、PULL_RR_JAM）がある。PULL_RR_JAMを指定されていて、バッファ装置のバッファ装置の識別子をBUFF_1、BUFF_2 ... BUFF_j ... BUFF_nとしてBUFF_jが輻輳していることを表す情報が状態管理部１４０に存在している場合、データ処理プログラム１５０はバッファ装置BUFF_jからのみデータ取得を試みる。
【００６７】
データ処理プログラム１５０は、データ処理装置１００上に１つ以上存在し、かつ、データ処理装置１００もシステム全体で１つ以上存在し、それぞれのデータ処理プログラム１５０がバッファ装置１０のデータを取得して処理を試みる。
【００６８】
本発明のシステムでは、PULL戦略の指定として複数のPULL戦略を指定することが可能である。たとえば、1つのデータ処理装置に2つのデータ処理プログラム（データ処理プログラムPG_Aとデータ処理プログラムPG_B）を起動し、PG_AにはPULL_RR、PG_BにはPULL_RANDといった個別の指定をすることが可能である。
【００６９】
PULL戦略が決定すると、データ処理プログラム１５０の起動が完了して（ステップ２０４）、データ処理フローのループ（図９）を開始する（ステップ２０５）。
【００７０】
図９は、本発明の第１の実施の形態におけるデータ処理装置でのデータ処理のフローチャートである。
【００７１】
データ処理プログラム１５０は、バッファ装置選択部１３０にバッファ装置の選択要求を行う（ステップ３０１）。バッファ装置選択部１３０は図１０に示す処理によりバッファ装置を選択し、データ処理プログラム１５０に返却する（ステップ３０２）。データ処理プログラム１５０は、バッファ装置選択結果を取得し（ステップ３０３）、図１１に示す処理を行なう（ステップ３０４）。
【００７２】
データ処理フローのループの中では、図９で示したように、バッファ装置選択部１３０がバッファ装置選択フロー（図１０）を行い、データ処理プログラム１５０がデータ取得と処理フロー（図１１）を順に実施する。
【００７３】
図１０は、本発明の第１の実施の形態におけるデータ処理装置でのバッファ装置選択のフローチャートである。
【００７４】
図１０で示したバッファ装置選択フローでは、バッファ装置選択部１３０がPULL戦略構成設定部１２０で取得したPULL戦略に応じて分岐を行う。例として、異なるPULL戦略で動作する2つのデータ処理プログラムPG_AとPG_Bが動作している例を示す。PG_Aには状態管理部がもつ情報に応じて動作を変更するPULL戦略1としてPULL_RR_JAMを指定する。PULL戦略2には状態管理部がもつ情報に関係なく動作するPULL_RANDを指定してする。それぞれのデータ処理プログラムは、指定されたPULL戦略で動作が分岐する（ステップ４０１）。
【００７５】
データ処理プログラムPG_Aは、PULL戦略1で動作することが指定されているため、バッファ装置選択が開始されると（ステップ４０２）、輻輳しているバッファ装置の存在を状態管理部１４０に確認して、バッファ装置リストを取得する（ステップ４０３）。バッファ装置リスト中に輻輳が発生しているバッファ装置の情報が存在していないことが分かると（ステップ４０４,No）、すべてのバッファ装置を対象にPULL_RRの手順でバッファ装置のうち１つをデータ取得先バッファ装置に選択する（ステップ４０５）。また輻輳が発生していることが分かると（ステップ４０４,Yes）輻輳しているバッファ装置のリストの中からデータ取得先バッファ装置を選択する（ステップ４０６）。
【００７６】
PG_Bは、PULL戦略2で動作することが指定されているため（ステップ４０１，４０７）、すべてのバッファ装置を対象にPULL_RRの手順でバッファ装置のうち１つをデータ取得先バッファ装置として選択する（ステップ４０８）。
【００７７】
バッファ装置選択部１３０は、バッファ装置選択結果をデータ処理プログラム１５０に返却する（ステップ４０９）。
【００７８】
図１１は、本発明の第１の実施の形態におけるデータ処理装置及びバッファ装置でのデータ取得処理のフローチャートである。
【００７９】
図１１のフローチャートでは、図１０に示す処理ですでにデータ取得先バッファ装置として選択したバッファ装置１０に対して、通信部１１０がデータ取得要求を行う（ステップ５０１）。バッファ装置１０は、返却するデータが存在する場合はデータを返却し（ステップ５０２，５０３）、データが存在しない場合は、データが存在しない旨の応答を返す（ステップ５０２，５０４）。データが返却され場合のみデータ処理プログラムはデータ処理を実行する（ステップ５０５，５０６）。データ処理については、システムの用途に応じて動作が異なるが、ここでは特に規定しない。
【００８０】
[第２の実施の形態]
本実施の形態では、明確にPULL戦略を分離していないが、バッファ装置選択部の中で条件分岐を行い、複数のPULL戦略が混在している状態の実装例を示す。
【００８１】
本実施の形態におけるデータ処理プログラムでは、輻輳が発生していない場合は、複数あるバッファ装置を決められたPULL戦略（例えば、ランダム、バッファ装置の識別子の値順など）で巡回しながらバッファ装置からデータを取得して処理を行う。一方、輻輳が発生している場合は、データ処理プログラムは、一つのシステム内の処理キャパシティのうち、αの割合で輻輳に対応する（０≦α≦１）。例えば、一つの具体例として、輻輳継続時間に基づいてアルファの値を大きくして段階的に輻輳対応のデータ処理プログラムを増やす方法や、また、輻輳レベル（例えば、輻輳しているバッファ装置の数）に基づいてαの値を大きくして、段階的に輻輳対応のデータ処理プログラムを増やすといった方法が適用可能である。
【００８２】
図１２は、本発明の第２の実施の形態におけるシステム構成を示す。
【００８３】
同図のシステムにおいて、バッファ装置１０は、第１の実施の形態と同様に、蓄積部１１、閾値情報１２、通信部１３を具備し、データ処理装置２００は、通信部２１０、バッファ装置選択部２３０、輻輳対応設定部２２０、状態管理部２４０、データ処理プログラム２５０からなる。またバッファ装置１０の前段には、データソース２０で発生したデータをバッファ装置１０に追記するプログラム（以下、追記クライアント２１）が存在することを前提とする。
【００８４】
各データ処理プログラムは、選択したPULL戦略に基づいて処理の度に、データを取得（フェッチ）するバッファ装置を選択して、データを取得する。この動作の流れを図１３〜図１７を用いて示す。
【００８５】
図１３は、本発明の第２の実施の形態におけるバッファ装置におけるデータ受信とデータ処理装置への輻輳状態の通知処理のフローチャートである。
【００８６】
バッファ装置１０は、外部の追記プログラム等から各データDが追記され、輻輳している場合には、データ処理装置２００に通知する。
【００８７】
図１３の開始は、前述の追記クライアントが通信部１３へデータ送信した時点からを記述してある。データDは、データソース２０で発生したストリーム処理の対象データであり、ある時刻tのデータDをDtと表す。
【００８８】
バッファ装置１０は、Dtを通信部１３で受信する（ステップ６０１）。通信部１３は、任意の通信プロトコルで受信可能であるが、通常はIPプロトコルが使用される。通信部１３が受信したDtは、一旦このバッファ装置１０の蓄積部１１にデータ保存されて、データ処理装置２００によって処理されるのを待つ（ステップ６０２，６０３）。当該データDtが到着する以前のデータDt-1や以後のデータDt+1と同様に受信したデータは、バッファ装置１０の蓄積部１１に保存される。このとき、蓄積部１１は、閾値情報を参照し（ステップ６０４）、その時点で蓄積部１１に蓄積されているデータの量が閾値情報を超えている場合は、輻輳通知としてデータ処理装置２００へ通知を行う（ステップ６０５，６０６）。
【００８９】
閾値情報の指定方法としては、追記されたデータDの数やデータDの容量の合計値などが利用可能である。
【００９０】
輻輳通知は、任意のプロトコルが使用可能であるが、通常はIPプロトコルが使用される。
【００９１】
さらに、輻輳通知は、バッファ装置１０のどの蓄積部１１が閾値を超えているかを一意に特定するための情報を含んでいる必要がある。例として、IPプロトコルを利用した実装においては、輻輳通知には、輻輳しているバッファ装置１０のIPアドレス、通信ポート番号を含んでいるため、データ処理装置２００は、どのバッファ装置が輻輳しているかを知ることができる。
【００９２】
また、別の例として、本システム内で、各バッファ装置１０を識別可能なバッファ装置ID情報とバッファ装置が動作しているIPアドレス、通信ポート番号との対応表を管理していて、バッファ装置ID情報のみを含め、各データ処理装置２００側でバッファ装置ID情報からIPアドレスと通信ポートを特定することでも、データ処理装置２００は、どのバッファ装置が輻輳しているかを知ることができる。
【００９３】
データ処理装置２００は、輻輳通知を受信すると、一旦輻輳しているバッファ装置の情報を状態管理部２４０に保存する。
【００９４】
データ処理装置２００は、状態管理部２４０および輻輳対応設定の内容に応じてバッファ装置１０からデータを取得してデータ処理を行う。この時のデータ処理装置２００におけるデータ処理を図１４〜図１７を用いて説明する。
【００９５】
図１４は、本発明の第２の実施の形態におけるデータ処理装置でのデータ処理全体のデータ処理のフローチャートである。
【００９６】
図１４の開始は、データ処理プログラム２５０の開始した時点から記述してある。データ処理プログラムが起動すると（ステップ７０１）（データ処理フローのループ（図１５のフロー）を開始する（ステップ７０２）。
【００９７】
データ処理フローのループの中では、図１５で示したように、データ処理プログラム２５０が、バッファ装置選択フロー（図１６）とデータ取得と処理フロー（図１７）を順に実施する。
【００９８】
図１６は、本発明の第２の実施の形態におけるデータ処理装置でのバッファ装置選択のフローチャートである。
【００９９】
バッファ装置選択部２３０では、状態管理部２４０からバッファ装置１０のリストを取得し（ステップ８０１）、この結果から輻輳が発生しているかどうか、および輻輳が発生しているバッファ装置を知ることができ、図１６で示したバッファ装置選択フローでは、この結果、輻輳が発生していない場合において、データ処理プログラム２５０は全バッファ装置を対象にデータ取得先バッファ装置を選択する。この時に実施するPULL戦略としては、上述したPULL_RANDやPULL_RRなど任意のPULL戦略を設定可能である。またバッファ装置１０において輻輳が発生している場合には（ステップ８０２、Yes）、さらに輻輳対応設定を取得し（ステップ８０３）、当該データ処理プログラムが輻輳に対応するべきかどうかを判断する（ステップ８０４）。この時、どの程度の割合で輻輳に対応すべきかを表すα値を調整する必要がある（ステップ８０５）。
【０１００】
調整方法の例としては、全体のバッファ装置数SUM_buffおよび、輻輳しているバッファ装置数SUM_jamからα＝SUM_jam／SUM_buffから求めるというものである。
【０１０１】
このαにより乱数判定では0〜1の間の乱数を発生し（ステップ８０６）、乱数が0以上α未満の場合は、データ処理プログラムは、輻輳状態のバッファ装置を対象にバッファ装置の選択を行う（ステップ８０７）。また、乱数がα以上1以下の場合は、データ処理プログラム２５０は、バッファ装置全体を対象にバッファ装置の選択を行う（ステップ８０８）。バッファ選択の結果をデータ処理プログラム２５０に出力する（ステップ８０９）。
【０１０２】
バッファ装置選択手順により、明示的に複数のPULL戦略を混在させなくても、状態管理部の状態に応じて輻輳に対応するPULL戦略と、輻輳に対応しないPULL戦略を混在させた状態と同様の機能を実現することができる。
【０１０３】
図１７は、本発明の第２の実施の形態におけるデータ処理装置でのデータ取得処理のフローチャートである。
【０１０４】
図１７の処理は、図１６の処理で既にデータ取得先バッファ装置として選択したバッファ装置１０に対して、通信部２１０がデータ取得要求を行う（ステップ９９０１）。バッファ装置１０は、返却するデータが存在する場合はデータを返却し（ステップ９０２）、データが存在しない場合は、データが存在しない旨の応答を返す（ステップ９０３）。データが返却され場合のみデータ処理プログラムはデータ処理を実行する（ステップ９０４）。データ処理については、システムの用途に応じて動作が異なるが、ここでは特に規定しない。
【０１０５】
以下に、本発明による効果を具体的に以下に示す。
【０１０６】
一つのケースとして、いくつものバッファ装置があり、一時的に大量のデータが到達して輻輳が発生する可能性があり優先的に処理する必要のあるバッファ装置BUFF_1と、流量は少ないが低レイテンシで処理を行う必要のあるバッファ装置BUFF_2が混在するシステムにおいて、PULL_RR（輻輳に対応しないPULL戦略）、PULL_RR_JAM（輻輳に対応するPULL戦略）を混在させた場合を例にあげる。
【０１０７】
輻輳が発生していない場合は、PULL_RRとPULL_RR_JAMは同様にラウンドロビンでバッファ装置を選択する。これによりすべてのバッファ装置が一定期間内に選択される。一部のバッファ装置にて輻輳が発生している場合は、PULL_RRで動作するデータ処理プログラムは、常に一定の周期でBUFF_1とBUFF_2を巡回するため、BUFF_2の要件を満たすことができる。また、PULL_RR_JAMは、輻輳したBUFF_1を優先的に選択するため、BUFF_1をに到着して蓄積されている大量のデータを短時間で処理しようと試みる。
【０１０８】
上述のように、単独のPULL戦略を選択した場合に、入力や処理の要件が異なるデータが混在する環境下で十分に要件を満たせない場合があり、このような環境下において、複数のPULL戦略を混在させることの可能な本発明のシステムの効果がある。
【０１０９】
上記の第１の実施の形態における図６のデータ処理装置１００、第２の実施の形態における図１２のデータ処理装置２００の構成要素の動作をプログラムとして構築し、ストリーム処理システムにおけるデータ処理装置として利用されるコンピュータにインストールして実行させる、または、ネットワークを介して流通させることが可能である。
【０１１０】
なお、本発明は、上記の第１、第２の実施の形態に限定されることなく、特許請求の範囲内において、種々変更・応用が可能である。
【符号の説明】
【０１１１】
１０  バッファ装置
１１  蓄積部
１２  閾値情報
１３  通信部
２０  データソース
２１  追記クライアント
１００，２００  データ処理装置
１１０，２１０  通信部
１２０  PULL戦略構成設定部
１３０，２３０  バッファ装置選択部
１４０，２４０  状態管理部
１５０，２５０  データ処理プログラム
２２０  輻輳対応設定部
【図１】
2014225111.tif 000003
【図２】
2014225111.tif 000004
【図３】
2014225111.tif 000005
【図４】
2014225111.tif 000006
【図５】
2014225111.tif 000007
【図６】
2014225111.tif 000008
【図７】
2014225111.tif 000009
【図８】
2014225111.tif 000010
【図９】
2014225111.tif 000011
【図１０】
2014225111.tif 000012
【図１１】
2014225111.tif 000013
【図１２】
2014225111.tif 000014
【図１３】
2014225111.tif 000015
【図１４】
2014225111.tif 000016
【図１５】
2014225111.tif 000017
【図１６】
2014225111.tif 000018
【図１７】
2014225111.tif 000019