トップに関する質問 Reactive Extensionsを使った一定間隔毎のイベントサンプリング方法

編集履歴

質問編集履歴

ソース修正

2016/05/14 07:45

投稿

hossy

スコア15

test CHANGED Viewed

File without changes

test CHANGED Viewed

@@ -338,7 +338,23 @@
         {
+            var typeA = datas.LastOrDefault(_ => _.SensorType == SensorType.TypeA);
+            var typeB = datas.LastOrDefault(_ => _.SensorType == SensorType.TypeB);
+            Debug.WriteLine(String.Format("{0} : Event : Id=[{1}], Count=[{2}], TypeA=[{3}], typeB=[{4}]",
+                DateTime.Now.ToString("HH:mm:ss"),
-...
+                grouped.Key,
+                grouped.Count(),
+                typeA?.Timestamp.ToString("HH:mm:ss"),
+                typeB?.Timestamp.ToString("HH:mm:ss")));
         }

記述補足

2016/05/14 07:45

投稿

hossy

スコア15

test CHANGED Viewed

File without changes

test CHANGED Viewed

@@ -256,7 +256,7 @@
-#内容補足
+#内容補足(追加1)

「更に試してみたこと」追加

2016/05/14 07:44

投稿

hossy

スコア15

test CHANGED Viewed

File without changes

test CHANGED Viewed

@@ -16,8 +16,6 @@
 - この処理をReactive Extensionsで書きたい
-ここに質問したいことを詳細に書いてください
 #書いてみたソースコード
@@ -308,163 +306,41 @@
-なお、A2とB1については、boundaryストリームのイベントタイミングで発火されれば、同一のイベントで渡されてきても、別のストリームで扱われてもどちらでも良いです。#やりたいことの概要
+なお、A2とB1については、boundaryストリームのイベントタイミングで発火されれば、同一のイベントで渡されてきても、別のイベントで扱われてもどちらでも良いです。
-Reactive Extensionsの書き方に関する質問です。
-やりたいことの概要は以下のようなものです。
-- イメージとしては、複数のセンサーを持つ機器(複数台)のイベントをサンプリングするIoTのようなもの
-- 機器はIdを持ち、センサーはSensorTypeによって識別され、その2つをキーとするイベントがストリームとして発生する
-- このストリームから、一定のサンプリング間隔で、機器(Id)毎に各センサーの直近の値を取得する
-- この処理をReactive Extensionsで書きたい
-ここに質問したいことを詳細に書いてください
-#書いてみたソースコード
+#更に試してみたこと(追加2)
-##データ構造
+更に以下のようなコードを書いてみました。
 ```cs
-// センサー種別
-public enum SensorType
-{
-    TypeA,
+observable
-    TypeB
-}
-// 発生するイベントのデータ構造
-public class DataEvent
-{
-    public int Id { get; set; }
-    public SensorType SensorType { get; set; }
+    .GroupBy(_ => new { _.Id, _.SensorType })
-    public DateTime Timestamp { get; set; }
-}
-```
-##イベントストリームの疑似
-```cs
-// イベントストリームの疑似Observable
+    .Select(_ => _.Sample(timeObservable))
+    .Merge()
-var observable = Observable.Create<DataEvent>(observer =>
+    .Buffer(timeObservable)
-{
-    int counter = 0;
+    .Subscribe(datas =>
-    return new System.Threading.Timer(state =>
     {
-        var data = new DataEvent { Id = counter / 2 % 2, SensorType = (SensorType)(counter % 2), Timestamp = DateTime.Now };
-        Debug.WriteLine(String.Format("{0} : Data  : Id=[{1}], Type=[{2}]",
-            data.Timestamp.ToString("HH:mm:ss"),
-            data.Id,
-            data.Type));
-        counter++;
-        observer.OnNext(data);
-    }, null, 0, 1000);
-});
-```
-##自分が作った処理
-```cs
-// 一定期間毎に、Id単位で直近の値をサンプリング
-var timeObservable = Observable.Interval(TimeSpan.FromSeconds(5)).Publish().RefCount();
-observable
-    .GroupBy(_ => _.Id)
+        foreach (var grouped in datas.GroupBy(_ => _.Id))
-    .Subscribe(groupedObservable =>
-    {
+        {
-        groupedObservable
-            .Buffer(timeObservable)
-            .Subscribe(datas =>
+...
-            {
+        }
-                var typeA = datas.LastOrDefault(_ => _.SensorType == SensorType.TypeA);
-                var typeB = datas.LastOrDefault(_ => _.SensorType == SensorType.TypeB);
-                Debug.WriteLine(String.Format("{0} : Event : Id=[{1}], Count=[{2}], TypeA=[{3}], typeB=[{4}]",
-                    DateTime.Now.ToString("HH:mm:ss"),
-                    groupedObservable.Key,
-                    datas.Count,
-                    typeA?.Timestamp.ToString("HH:mm:ss"),
-                    typeB?.Timestamp.ToString("HH:mm:ss")));
-            });
     });
@@ -472,150 +348,6 @@
-#実行例
-##出力
-```
-16:57:05 : Data  : Id=[0], Type=[TypeA]
-16:57:06 : Data  : Id=[0], Type=[TypeB]
-16:57:07 : Data  : Id=[1], Type=[TypeA]
-16:57:08 : Data  : Id=[1], Type=[TypeB]
-16:57:09 : Data  : Id=[0], Type=[TypeA]
-16:57:10 : Event : Id=[0], Count=[3], TypeA=[16:57:09], typeB=[16:57:06]
-16:57:10 : Event : Id=[1], Count=[2], TypeA=[16:57:07], typeB=[16:57:08]
-16:57:10 : Data  : Id=[0], Type=[TypeB]
-16:57:11 : Data  : Id=[1], Type=[TypeA]
-16:57:12 : Data  : Id=[1], Type=[TypeB]
-16:57:13 : Data  : Id=[0], Type=[TypeA]
-16:57:14 : Data  : Id=[0], Type=[TypeB]
-16:57:15 : Event : Id=[0], Count=[3], TypeA=[16:57:13], typeB=[16:57:14]
-16:57:15 : Event : Id=[1], Count=[2], TypeA=[16:57:11], typeB=[16:57:12]
-16:57:15 : Data  : Id=[1], Type=[TypeA]
-16:57:16 : Data  : Id=[1], Type=[TypeB]
-16:57:17 : Data  : Id=[0], Type=[TypeA]
-16:57:18 : Data  : Id=[0], Type=[TypeB]
-16:57:19 : Data  : Id=[1], Type=[TypeA]
-16:57:20 : Event : Id=[0], Count=[2], TypeA=[16:57:17], typeB=[16:57:18]
-16:57:20 : Event : Id=[1], Count=[3], TypeA=[16:57:19], typeB=[16:57:16]
-```
-##解説
-```
-1 16:57:05 : Data  : Id=[0], Type=[TypeA]
-2 16:57:06 : Data  : Id=[0], Type=[TypeB]
-3 16:57:07 : Data  : Id=[1], Type=[TypeA]
-4 16:57:08 : Data  : Id=[1], Type=[TypeB]
-5 16:57:09 : Data  : Id=[0], Type=[TypeA]
-6 16:57:10 : Event : Id=[0], Count=[3], TypeA=[16:57:09], typeB=[16:57:06] 5と2が直近
-7 16:57:10 : Event : Id=[1], Count=[2], TypeA=[16:57:07], typeB=[16:57:08] 3と4が直近
-```
-#質問事項
-とりあえずやりたいことは出来たのですが、Reactive Extensionsを使った書き方でもっと効率の良い書き方はできるのでしょうか？
+このコードはバッファするオブジェクト自体は最小になっており、惜しいのですが致命的な問題がある認識です。
-例えば、この方法だとサンプリング間隔間のデータはすべてListで保持され、Subscribe()の中でLastOrDefault()で直近の値を取得していますが、Take()等によりストリームの段階で直近の1件のみを処理するようにする等。
-それも試してみようと思ったのですが、Take()したいのはIdとSensorTypeの復号キー単位だが最終的に取得したいのはId単位な点、また、その場合にSample()等をどのタイミングで呼び出すのかがわからなくなり、ギブアップしました。
+本来は同じものとして処理したいtimeObservableを、キー毎のSample()用とイベント発火間隔Buffer()で個別に処理しているため、厳密な処理にならず、そのためか、この処理だとtimeObservableが5秒間隔に対して、初回イベントが10秒後に発生しその後は5秒間隔で発生する、っというような動きになりました。
-#内容補足
-自分のやりたい内容を、以下に整理してみました。
-```
-source   |-A1-A2-B1----B2----A3-A4-B3-B4---
-boundary |----------X-----Y--------------Z-
-                    A2    B2             A4
-                    B1                   B4
-```
-sourceストームのA、B等がキーで、boundaryストリームのイベント発生のタイミングで同じキーの最後の値のみを取得したいのです。
-現状出来ている内容だと以下のような内容で、イベント発生後に1A、1C、1Dを捨てる形になるので、boundaryのイベント発生間隔が長くなると、その間sourceストリームで発生したオブジェクト全てが保持されてしまいます。
-```
-source   |-A1-A2-B1----B2----A3-A4-B3-B4---
-boundary |----------X-----Y--------------Z-
-                    A1    B2             A3
-                    A2                   A4
-                    B1                   B3
-                                         B4
-```
-なお、A2とB1については、boundaryストリームのイベントタイミングで発火されれば、同一のイベントで渡されてきても、別のイベントで扱われてもどちらでも良いです。

文法の修正

2016/05/14 07:43

投稿

hossy

スコア15

test CHANGED Viewed

File without changes

test CHANGED Viewed

@@ -618,4 +618,4 @@
-なお、A2とB1については、boundaryストリームのイベントタイミングで発火されれば、同一のイベントで渡されてきても、別のストリームで扱われてもどちらでも良いです。
+なお、A2とB1については、boundaryストリームのイベントタイミングで発火されれば、同一のイベントで渡されてきても、別のイベントで扱われてもどちらでも良いです。

サンプル誤り修正

2016/05/14 05:14

投稿

hossy

スコア15

test CHANGED Viewed

File without changes

test CHANGED Viewed

@@ -292,17 +292,327 @@
 ```
-source   |-1A-2A-1B----2B----1C-1D-2C-2D---
+source   |-A1-A2-B1----B2----A3-A4-B3-B4---
 boundary |----------X-----Y--------------Z-
-                    1A    2B             1C
+                    A1    B2             A3
-                    2A                   1D
+                    A2                   A4
-                    1B                   2C
+                    B1                   B3
-                                         2D
+                                         B4
+```
+なお、A2とB1については、boundaryストリームのイベントタイミングで発火されれば、同一のイベントで渡されてきても、別のストリームで扱われてもどちらでも良いです。#やりたいことの概要
+Reactive Extensionsの書き方に関する質問です。
+やりたいことの概要は以下のようなものです。
+- イメージとしては、複数のセンサーを持つ機器(複数台)のイベントをサンプリングするIoTのようなもの
+- 機器はIdを持ち、センサーはSensorTypeによって識別され、その2つをキーとするイベントがストリームとして発生する
+- このストリームから、一定のサンプリング間隔で、機器(Id)毎に各センサーの直近の値を取得する
+- この処理をReactive Extensionsで書きたい
+ここに質問したいことを詳細に書いてください
+#書いてみたソースコード
+##データ構造
+```cs
+// センサー種別
+public enum SensorType
+{
+    TypeA,
+    TypeB
+}
+// 発生するイベントのデータ構造
+public class DataEvent
+{
+    public int Id { get; set; }
+    public SensorType SensorType { get; set; }
+    public DateTime Timestamp { get; set; }
+}
+```
+##イベントストリームの疑似
+```cs
+// イベントストリームの疑似Observable
+var observable = Observable.Create<DataEvent>(observer =>
+{
+    int counter = 0;
+    return new System.Threading.Timer(state =>
+    {
+        var data = new DataEvent { Id = counter / 2 % 2, SensorType = (SensorType)(counter % 2), Timestamp = DateTime.Now };
+        Debug.WriteLine(String.Format("{0} : Data  : Id=[{1}], Type=[{2}]",
+            data.Timestamp.ToString("HH:mm:ss"),
+            data.Id,
+            data.Type));
+        counter++;
+        observer.OnNext(data);
+    }, null, 0, 1000);
+});
+```
+##自分が作った処理
+```cs
+// 一定期間毎に、Id単位で直近の値をサンプリング
+var timeObservable = Observable.Interval(TimeSpan.FromSeconds(5)).Publish().RefCount();
+observable
+    .GroupBy(_ => _.Id)
+    .Subscribe(groupedObservable =>
+    {
+        groupedObservable
+            .Buffer(timeObservable)
+            .Subscribe(datas =>
+            {
+                var typeA = datas.LastOrDefault(_ => _.SensorType == SensorType.TypeA);
+                var typeB = datas.LastOrDefault(_ => _.SensorType == SensorType.TypeB);
+                Debug.WriteLine(String.Format("{0} : Event : Id=[{1}], Count=[{2}], TypeA=[{3}], typeB=[{4}]",
+                    DateTime.Now.ToString("HH:mm:ss"),
+                    groupedObservable.Key,
+                    datas.Count,
+                    typeA?.Timestamp.ToString("HH:mm:ss"),
+                    typeB?.Timestamp.ToString("HH:mm:ss")));
+            });
+    });
+```
+#実行例
+##出力
+```
+16:57:05 : Data  : Id=[0], Type=[TypeA]
+16:57:06 : Data  : Id=[0], Type=[TypeB]
+16:57:07 : Data  : Id=[1], Type=[TypeA]
+16:57:08 : Data  : Id=[1], Type=[TypeB]
+16:57:09 : Data  : Id=[0], Type=[TypeA]
+16:57:10 : Event : Id=[0], Count=[3], TypeA=[16:57:09], typeB=[16:57:06]
+16:57:10 : Event : Id=[1], Count=[2], TypeA=[16:57:07], typeB=[16:57:08]
+16:57:10 : Data  : Id=[0], Type=[TypeB]
+16:57:11 : Data  : Id=[1], Type=[TypeA]
+16:57:12 : Data  : Id=[1], Type=[TypeB]
+16:57:13 : Data  : Id=[0], Type=[TypeA]
+16:57:14 : Data  : Id=[0], Type=[TypeB]
+16:57:15 : Event : Id=[0], Count=[3], TypeA=[16:57:13], typeB=[16:57:14]
+16:57:15 : Event : Id=[1], Count=[2], TypeA=[16:57:11], typeB=[16:57:12]
+16:57:15 : Data  : Id=[1], Type=[TypeA]
+16:57:16 : Data  : Id=[1], Type=[TypeB]
+16:57:17 : Data  : Id=[0], Type=[TypeA]
+16:57:18 : Data  : Id=[0], Type=[TypeB]
+16:57:19 : Data  : Id=[1], Type=[TypeA]
+16:57:20 : Event : Id=[0], Count=[2], TypeA=[16:57:17], typeB=[16:57:18]
+16:57:20 : Event : Id=[1], Count=[3], TypeA=[16:57:19], typeB=[16:57:16]
+```
+##解説
+```
+1 16:57:05 : Data  : Id=[0], Type=[TypeA]
+2 16:57:06 : Data  : Id=[0], Type=[TypeB]
+3 16:57:07 : Data  : Id=[1], Type=[TypeA]
+4 16:57:08 : Data  : Id=[1], Type=[TypeB]
+5 16:57:09 : Data  : Id=[0], Type=[TypeA]
+6 16:57:10 : Event : Id=[0], Count=[3], TypeA=[16:57:09], typeB=[16:57:06] 5と2が直近
+7 16:57:10 : Event : Id=[1], Count=[2], TypeA=[16:57:07], typeB=[16:57:08] 3と4が直近
+```
+#質問事項
+とりあえずやりたいことは出来たのですが、Reactive Extensionsを使った書き方でもっと効率の良い書き方はできるのでしょうか？
+例えば、この方法だとサンプリング間隔間のデータはすべてListで保持され、Subscribe()の中でLastOrDefault()で直近の値を取得していますが、Take()等によりストリームの段階で直近の1件のみを処理するようにする等。
+それも試してみようと思ったのですが、Take()したいのはIdとSensorTypeの復号キー単位だが最終的に取得したいのはId単位な点、また、その場合にSample()等をどのタイミングで呼び出すのかがわからなくなり、ギブアップしました。
+#内容補足
+自分のやりたい内容を、以下に整理してみました。
+```
+source   |-A1-A2-B1----B2----A3-A4-B3-B4---
+boundary |----------X-----Y--------------Z-
+                    A2    B2             A4
+                    B1                   B4
+```
+sourceストームのA、B等がキーで、boundaryストリームのイベント発生のタイミングで同じキーの最後の値のみを取得したいのです。
+現状出来ている内容だと以下のような内容で、イベント発生後に1A、1C、1Dを捨てる形になるので、boundaryのイベント発生間隔が長くなると、その間sourceストリームで発生したオブジェクト全てが保持されてしまいます。
+```
+source   |-A1-A2-B1----B2----A3-A4-B3-B4---
+boundary |----------X-----Y--------------Z-
+                    A1    B2             A3
+                    A2                   A4
+                    B1                   B3
+                                         B4
 ```

補足の内容をわかりやすく修正

2016/05/14 05:13

投稿

hossy

スコア15

test CHANGED Viewed

File without changes

test CHANGED Viewed

@@ -270,7 +270,7 @@
 source   |-A1-A2-B1----B2----A3-A4-B3-B4---
-boundary |----------0-----1--------------3-
+boundary |----------X-----Y--------------Z-
                     A2    B2             A4
@@ -294,7 +294,7 @@
 source   |-1A-2A-1B----2B----1C-1D-2C-2D---
-boundary |----------0-----1--------------3-
+boundary |----------X-----Y--------------Z-
                     1A    2B             1C

内容補足追加

2016/05/14 05:11

投稿

hossy

スコア15

test CHANGED Viewed

File without changes

test CHANGED Viewed

@@ -255,3 +255,57 @@
 例えば、この方法だとサンプリング間隔間のデータはすべてListで保持され、Subscribe()の中でLastOrDefault()で直近の値を取得していますが、Take()等によりストリームの段階で直近の1件のみを処理するようにする等。
 それも試してみようと思ったのですが、Take()したいのはIdとSensorTypeの復号キー単位だが最終的に取得したいのはId単位な点、また、その場合にSample()等をどのタイミングで呼び出すのかがわからなくなり、ギブアップしました。
+#内容補足
+自分のやりたい内容を、以下に整理してみました。
+```
+source   |-A1-A2-B1----B2----A3-A4-B3-B4---
+boundary |----------0-----1--------------3-
+                    A2    B2             A4
+                    B1                   B4
+```
+sourceストームのA、B等がキーで、boundaryストリームのイベント発生のタイミングで同じキーの最後の値のみを取得したいのです。
+現状出来ている内容だと以下のような内容で、イベント発生後に1A、1C、1Dを捨てる形になるので、boundaryのイベント発生間隔が長くなると、その間sourceストリームで発生したオブジェクト全てが保持されてしまいます。
+```
+source   |-1A-2A-1B----2B----1C-1D-2C-2D---
+boundary |----------0-----1--------------3-
+                    1A    2B             1C
+                    2A                   1D
+                    1B                   2C
+                                         2D
+```
+なお、A2とB1については、boundaryストリームのイベントタイミングで発火されれば、同一のイベントで渡されてきても、別のストリームで扱われてもどちらでも良いです。