You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2021/12/06 19:49:46 UTC
[GitHub] [arrow] zixi-bwang opened a new pull request #11874: A Real-time C# memory-based data analytics example
zixi-bwang opened a new pull request #11874:
URL: https://github.com/apache/arrow/pull/11874
A Real-time C# memory-based data analytics platform, which leverages Apache Arrow as the unified data store.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] eerhardt commented on pull request #11874: MINOR: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
eerhardt commented on pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#issuecomment-987159057
One quick question/suggestion: Can you add a small `README` to the example explaining what it is showing? Describe the scenario, etc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] eerhardt commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
eerhardt commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r771593521
##########
File path: csharp/examples/IoTExample/IoTExample.csproj
##########
@@ -0,0 +1,16 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+ <PropertyGroup>
+ <OutputType>Exe</OutputType>
+ <TargetFramework>netcoreapp2.1</TargetFramework>
Review comment:
```suggestion
<TargetFramework>netcoreapp3.1</TargetFramework>
```
2.1 is EOL'd. Can you update this to 3.1?
##########
File path: csharp/examples/IoTExample/README.md
##########
@@ -0,0 +1,29 @@
+# .NET IoT Analytics
+
+## Introduction
+
+For Big Data problems, people will find that most of the technology and resources are for Java, Python, R, and Scala.
+Unfortunately, it's not the same for C#. But with Apache Arrow, a language-agnostic data format, the gap between Big Data Technology and
+.Net Enterprise Software Development can be bridged.
+
+## The original dataset is:
+
+[WISDM Smartphone and Smartwatch Activity and Biometrics Dataset Dataset](https://archive.ics.uci.edu/ml/datasets/WISDM+Smartphone+and+Smartwatch+Activity+and+Biometrics+Dataset+):
+ Contains accelerometer and gyroscope time-series sensor data collected from a smartphone and smartwatch as 51 test subjects perform 18 activities for 3 minutes each.
+
+## The sample dataset used in this example is randomly generated in order to test arrow in large-scale data scenario
+
+The sample dataset includes activity data from 1000 participants from an activity recognition project.
+Each participant performed each of the 18 activities for a total amount of one billion accelerometer data events
+reported from smartphone and smartwatch.
+
+Timestamp is the time at which the sensor reported the reading.
+X_Axis is the g-force acceleration along the x-axis.
+Y_Axis is the g-force acceleration along the y-axis.
+Z_Axis is the g-force acceleration along the z-axis.
Review comment:
```suggestion
* Timestamp is the time at which the sensor reported the reading.
* X_Axis is the g-force acceleration along the x-axis.
* Y_Axis is the g-force acceleration along the y-axis.
* Z_Axis is the g-force acceleration along the z-axis.
```
It may make this easier to read with bulleting.
##########
File path: csharp/examples/IoTExample/Model/SampleDataset.cs
##########
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+
+namespace IoTExample
+{
+ public class SampleDataset
+ {
+ private int _size;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly BlockingCollection<SensorData> _rows;
+
+ private readonly List<int> _colSubjectId;
+ private readonly List<string> _colActivityLabel;
+ private readonly List<long> _colTimestamp;
+ private readonly List<double> _colXAxis;
+ private readonly List<double> _colYAxis;
+ private readonly List<double> _colZAxis;
+
+ private readonly int _threshold = 1_000_000;
+ private readonly List<RecordBatch> _recordBatches;
+ private readonly MemoryAllocator _memoryAllocator;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataset(int inputs, int capacity)
+ {
+ _inputs = inputs;
+ _capacity = capacity;
+ _rows = new BlockingCollection<SensorData>(_capacity);
+
+ _colSubjectId = new List<int>(capacity);
+ _colActivityLabel = new List<string>();
+ _colTimestamp = new List<long>();
+ _colXAxis = new List<double>();
+ _colYAxis = new List<double>();
+ _colZAxis = new List<double>();
+
+ _recordBatches = new List<RecordBatch>();
+ _memoryAllocator = new NativeMemoryAllocator(alignment: 64);
+ }
+
+ public void Produce()
+ {
+ Random rand = new Random();
+ bool success;
+
+ List<string> keyList = new List<string>(activityLabel.Keys);
+ int count = keyList.Count;
+
+ DateTime now = DateTime.Now;
+ long unixTime = ((DateTimeOffset)now).ToUnixTimeSeconds();
+
+ while (_size < _inputs)
+ {
+ string randomKey = keyList[rand.Next(count)];
+ string label = activityLabel[randomKey];
+
+ // generate missing values
+ if (rand.Next(10_000) == 9_999)
+ {
+ label = null;
+ }
+
+ success = _rows.TryAdd(new SensorData
+ {
+ subjectId = rand.Next(1000, 2001),
+ activityLabel = label,
+ timestamp = unixTime++,
+ x_Axis = rand.NextDouble(),
+ y_Axis = rand.NextDouble(),
+ z_Axis = rand.NextDouble(),
+ });
+
+ if (success)
+ {
+ //Console.WriteLine($"Enqueue Task 0");
+ _size++;
+ }
+ else
+ {
+ Console.WriteLine("Producing is blocked, percent completed is: {0}%", Math.Round((double)_size / _inputs, 4) * 100);
+ }
+ }
+
+ _rows.CompleteAdding();
+ }
+
+ public async void Consume()
+ {
+ while (!_rows.IsCompleted)
+ {
+ if (!_rows.TryTake(out SensorData item, 3000))
+ {
+ Console.WriteLine("Consuming is blocked!");
+ }
+ else
+ {
+ //Console.WriteLine($"Dequeue Task 1");
+ if (item != null && item.subjectId != null)
+ {
+ _colSubjectId.Add((int)item.subjectId);
+ _colActivityLabel.Add(item.activityLabel);
+ _colTimestamp.Add((long)item.timestamp);
+ _colXAxis.Add((double)item.x_Axis);
+ _colYAxis.Add((double)item.y_Axis);
+ _colZAxis.Add((double)item.z_Axis);
+ }
+ }
+
+ if (_colSubjectId.Count == _threshold)
+ {
+ // Build a record batch using the Fluent API
+ var recordBatch = new RecordBatch.Builder(_memoryAllocator)
+ .Append("SubjectId", false, col => col.Int32(array => array.AppendRange(_colSubjectId)))
+ .Append("ActivityLabel", false, col => col.String(array => array.AppendRange(_colActivityLabel)))
+ .Append("Timestamp", false, col => col.Int64(array => array.AppendRange(_colTimestamp)))
+ .Append("XAxis", false, col => col.Double(array => array.AppendRange(_colXAxis)))
+ .Append("YAxis", false, col => col.Double(array => array.AppendRange(_colYAxis)))
+ .Append("ZAxis", false, col => col.Double(array => array.AppendRange(_colZAxis)))
+ .Build();
+
+ _recordBatches.Add(recordBatch);
+
+ _colSubjectId.Clear();
+ _colActivityLabel.Clear();
+ _colTimestamp.Clear();
+ _colXAxis.Clear();
+ _colYAxis.Clear();
+ _colZAxis.Clear();
+ }
+ }
+ }
+
+ public async Task<bool> PersistData()
+ {
+ string time = DateTime.Now.ToString("yyyyMMdd_HHmmss");
+ Stopwatch stopwatch = new Stopwatch();
+ stopwatch.Start();
+
+ Schema schema = _recordBatches[0].Schema;
+ // Write record batch to a file
+ using (var stream = File.OpenWrite(@"c:\temp\data\iotbigdata.arrow"))
Review comment:
If this path doesn't exist, you get an error at the end of the run:
```
Producing is blocked, percent completed is: 90%
Unhandled exception. System.IO.DirectoryNotFoundException: Could not find a part of the path 'c:\temp\data\iotbigdata.arrow'.
at System.IO.FileStream.ValidateFileHandle(SafeFileHandle fileHandle)
at System.IO.FileStream.CreateFileOpenHandle(FileMode mode, FileShare share, FileOptions options)
at System.IO.FileStream..ctor(String path, FileMode mode, FileAccess access, FileShare share, Int32 bufferSize, FileOptions options)
at System.IO.FileStream..ctor(String path, FileMode mode, FileAccess access, FileShare share)
at System.IO.File.OpenWrite(String path)
at IoTExample.SampleDataset.PersistData() in C:\git\arrow\csharp\examples\IoTExample\Model\SampleDataset.cs:line 185
at IoTExample.Program.Main(String[] args) in C:\git\arrow\csharp\examples\IoTExample\Program.cs:line 42
at IoTExample.Program.<Main>(String[] args)
```
Maybe just write the file to the current directory?
##########
File path: csharp/examples/IoTExample/Program.cs
##########
@@ -0,0 +1,71 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using Apache.Arrow.Ipc;
+
+namespace IoTExample
+{
+ class Program
+ {
+ public static int concurrencyLevel = 1;
+ public static int totalInputs = 10_000_000;
+ public static int queueCapacity = 1_000_000;
+
+ public static async Task Main(string[] args)
+ {
+ SampleDataset sd = new SampleDataset(totalInputs, queueCapacity);
+
+ Console.WriteLine("Receiving IoT data...");
+ Task t1 = Task.Run(() => sd.Produce());
+
+ Console.WriteLine("Transforming data...");
+ Task t2 = Task.Run(() => sd.Consume());
+
+ // Wait for all tasks to complete
+ Task.WaitAll(t1, t2);
+
+ var success = await sd.PersistData();
+
+ if (!success)
+ return;
+
+ Console.WriteLine("Reading arrow files...");
+ var stream = File.OpenRead(@"c:\temp\data\iotbigdata.arrow");
Review comment:
Since this path needs to be in sync with `PersistData`, it may make sense to declare it in one spot, and pass it into `SampleDataset`.
##########
File path: csharp/examples/IoTExample/Model/SampleDataset.cs
##########
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+
+namespace IoTExample
+{
+ public class SampleDataset
+ {
+ private int _size;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly BlockingCollection<SensorData> _rows;
+
+ private readonly List<int> _colSubjectId;
+ private readonly List<string> _colActivityLabel;
+ private readonly List<long> _colTimestamp;
+ private readonly List<double> _colXAxis;
+ private readonly List<double> _colYAxis;
+ private readonly List<double> _colZAxis;
+
+ private readonly int _threshold = 1_000_000;
+ private readonly List<RecordBatch> _recordBatches;
+ private readonly MemoryAllocator _memoryAllocator;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataset(int inputs, int capacity)
+ {
+ _inputs = inputs;
+ _capacity = capacity;
+ _rows = new BlockingCollection<SensorData>(_capacity);
+
+ _colSubjectId = new List<int>(capacity);
+ _colActivityLabel = new List<string>();
+ _colTimestamp = new List<long>();
+ _colXAxis = new List<double>();
+ _colYAxis = new List<double>();
+ _colZAxis = new List<double>();
+
+ _recordBatches = new List<RecordBatch>();
+ _memoryAllocator = new NativeMemoryAllocator(alignment: 64);
+ }
+
+ public void Produce()
+ {
+ Random rand = new Random();
+ bool success;
+
+ List<string> keyList = new List<string>(activityLabel.Keys);
+ int count = keyList.Count;
+
+ DateTime now = DateTime.Now;
+ long unixTime = ((DateTimeOffset)now).ToUnixTimeSeconds();
+
+ while (_size < _inputs)
+ {
+ string randomKey = keyList[rand.Next(count)];
+ string label = activityLabel[randomKey];
+
+ // generate missing values
+ if (rand.Next(10_000) == 9_999)
+ {
+ label = null;
+ }
+
+ success = _rows.TryAdd(new SensorData
+ {
+ subjectId = rand.Next(1000, 2001),
+ activityLabel = label,
+ timestamp = unixTime++,
+ x_Axis = rand.NextDouble(),
+ y_Axis = rand.NextDouble(),
+ z_Axis = rand.NextDouble(),
+ });
+
+ if (success)
+ {
+ //Console.WriteLine($"Enqueue Task 0");
+ _size++;
+ }
+ else
+ {
+ Console.WriteLine("Producing is blocked, percent completed is: {0}%", Math.Round((double)_size / _inputs, 4) * 100);
+ }
+ }
+
+ _rows.CompleteAdding();
+ }
+
+ public async void Consume()
+ {
+ while (!_rows.IsCompleted)
+ {
+ if (!_rows.TryTake(out SensorData item, 3000))
+ {
+ Console.WriteLine("Consuming is blocked!");
+ }
+ else
+ {
+ //Console.WriteLine($"Dequeue Task 1");
+ if (item != null && item.subjectId != null)
+ {
+ _colSubjectId.Add((int)item.subjectId);
+ _colActivityLabel.Add(item.activityLabel);
+ _colTimestamp.Add((long)item.timestamp);
+ _colXAxis.Add((double)item.x_Axis);
+ _colYAxis.Add((double)item.y_Axis);
+ _colZAxis.Add((double)item.z_Axis);
+ }
+ }
+
+ if (_colSubjectId.Count == _threshold)
+ {
+ // Build a record batch using the Fluent API
+ var recordBatch = new RecordBatch.Builder(_memoryAllocator)
+ .Append("SubjectId", false, col => col.Int32(array => array.AppendRange(_colSubjectId)))
+ .Append("ActivityLabel", false, col => col.String(array => array.AppendRange(_colActivityLabel)))
+ .Append("Timestamp", false, col => col.Int64(array => array.AppendRange(_colTimestamp)))
+ .Append("XAxis", false, col => col.Double(array => array.AppendRange(_colXAxis)))
+ .Append("YAxis", false, col => col.Double(array => array.AppendRange(_colYAxis)))
+ .Append("ZAxis", false, col => col.Double(array => array.AppendRange(_colZAxis)))
+ .Build();
+
+ _recordBatches.Add(recordBatch);
+
+ _colSubjectId.Clear();
+ _colActivityLabel.Clear();
+ _colTimestamp.Clear();
+ _colXAxis.Clear();
+ _colYAxis.Clear();
+ _colZAxis.Clear();
+ }
+ }
+ }
+
+ public async Task<bool> PersistData()
Review comment:
```suggestion
public async Task PersistData()
```
This method never returns `false`. So you can just change it to return `Task` instead.
##########
File path: csharp/examples/IoTExample/Model/SampleDataset.cs
##########
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+
+namespace IoTExample
+{
+ public class SampleDataset
+ {
+ private int _size;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly BlockingCollection<SensorData> _rows;
+
+ private readonly List<int> _colSubjectId;
+ private readonly List<string> _colActivityLabel;
+ private readonly List<long> _colTimestamp;
+ private readonly List<double> _colXAxis;
+ private readonly List<double> _colYAxis;
+ private readonly List<double> _colZAxis;
+
+ private readonly int _threshold = 1_000_000;
+ private readonly List<RecordBatch> _recordBatches;
+ private readonly MemoryAllocator _memoryAllocator;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataset(int inputs, int capacity)
+ {
+ _inputs = inputs;
+ _capacity = capacity;
+ _rows = new BlockingCollection<SensorData>(_capacity);
+
+ _colSubjectId = new List<int>(capacity);
+ _colActivityLabel = new List<string>();
+ _colTimestamp = new List<long>();
+ _colXAxis = new List<double>();
+ _colYAxis = new List<double>();
+ _colZAxis = new List<double>();
+
+ _recordBatches = new List<RecordBatch>();
+ _memoryAllocator = new NativeMemoryAllocator(alignment: 64);
+ }
+
+ public void Produce()
+ {
+ Random rand = new Random();
+ bool success;
+
+ List<string> keyList = new List<string>(activityLabel.Keys);
+ int count = keyList.Count;
+
+ DateTime now = DateTime.Now;
+ long unixTime = ((DateTimeOffset)now).ToUnixTimeSeconds();
+
+ while (_size < _inputs)
+ {
+ string randomKey = keyList[rand.Next(count)];
+ string label = activityLabel[randomKey];
+
+ // generate missing values
+ if (rand.Next(10_000) == 9_999)
+ {
+ label = null;
+ }
+
+ success = _rows.TryAdd(new SensorData
+ {
+ subjectId = rand.Next(1000, 2001),
+ activityLabel = label,
+ timestamp = unixTime++,
+ x_Axis = rand.NextDouble(),
+ y_Axis = rand.NextDouble(),
+ z_Axis = rand.NextDouble(),
+ });
+
+ if (success)
+ {
+ //Console.WriteLine($"Enqueue Task 0");
+ _size++;
+ }
+ else
+ {
+ Console.WriteLine("Producing is blocked, percent completed is: {0}%", Math.Round((double)_size / _inputs, 4) * 100);
+ }
+ }
+
+ _rows.CompleteAdding();
+ }
+
+ public async void Consume()
+ {
+ while (!_rows.IsCompleted)
+ {
+ if (!_rows.TryTake(out SensorData item, 3000))
+ {
+ Console.WriteLine("Consuming is blocked!");
+ }
+ else
+ {
+ //Console.WriteLine($"Dequeue Task 1");
Review comment:
```suggestion
```
##########
File path: csharp/examples/IoTExample/Model/SampleDataset.cs
##########
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+
+namespace IoTExample
+{
+ public class SampleDataset
+ {
+ private int _size;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly BlockingCollection<SensorData> _rows;
+
+ private readonly List<int> _colSubjectId;
+ private readonly List<string> _colActivityLabel;
+ private readonly List<long> _colTimestamp;
+ private readonly List<double> _colXAxis;
+ private readonly List<double> _colYAxis;
+ private readonly List<double> _colZAxis;
+
+ private readonly int _threshold = 1_000_000;
+ private readonly List<RecordBatch> _recordBatches;
+ private readonly MemoryAllocator _memoryAllocator;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataset(int inputs, int capacity)
+ {
+ _inputs = inputs;
+ _capacity = capacity;
+ _rows = new BlockingCollection<SensorData>(_capacity);
+
+ _colSubjectId = new List<int>(capacity);
+ _colActivityLabel = new List<string>();
+ _colTimestamp = new List<long>();
+ _colXAxis = new List<double>();
+ _colYAxis = new List<double>();
+ _colZAxis = new List<double>();
+
+ _recordBatches = new List<RecordBatch>();
+ _memoryAllocator = new NativeMemoryAllocator(alignment: 64);
+ }
+
+ public void Produce()
+ {
+ Random rand = new Random();
+ bool success;
+
+ List<string> keyList = new List<string>(activityLabel.Keys);
+ int count = keyList.Count;
+
+ DateTime now = DateTime.Now;
+ long unixTime = ((DateTimeOffset)now).ToUnixTimeSeconds();
+
+ while (_size < _inputs)
+ {
+ string randomKey = keyList[rand.Next(count)];
+ string label = activityLabel[randomKey];
+
+ // generate missing values
+ if (rand.Next(10_000) == 9_999)
+ {
+ label = null;
+ }
+
+ success = _rows.TryAdd(new SensorData
+ {
+ subjectId = rand.Next(1000, 2001),
+ activityLabel = label,
+ timestamp = unixTime++,
+ x_Axis = rand.NextDouble(),
+ y_Axis = rand.NextDouble(),
+ z_Axis = rand.NextDouble(),
+ });
+
+ if (success)
+ {
+ //Console.WriteLine($"Enqueue Task 0");
+ _size++;
+ }
+ else
+ {
+ Console.WriteLine("Producing is blocked, percent completed is: {0}%", Math.Round((double)_size / _inputs, 4) * 100);
+ }
+ }
+
+ _rows.CompleteAdding();
+ }
+
+ public async void Consume()
+ {
+ while (!_rows.IsCompleted)
+ {
+ if (!_rows.TryTake(out SensorData item, 3000))
+ {
+ Console.WriteLine("Consuming is blocked!");
+ }
+ else
+ {
+ //Console.WriteLine($"Dequeue Task 1");
+ if (item != null && item.subjectId != null)
+ {
+ _colSubjectId.Add((int)item.subjectId);
+ _colActivityLabel.Add(item.activityLabel);
+ _colTimestamp.Add((long)item.timestamp);
+ _colXAxis.Add((double)item.x_Axis);
+ _colYAxis.Add((double)item.y_Axis);
+ _colZAxis.Add((double)item.z_Axis);
+ }
+ }
+
+ if (_colSubjectId.Count == _threshold)
+ {
+ // Build a record batch using the Fluent API
+ var recordBatch = new RecordBatch.Builder(_memoryAllocator)
+ .Append("SubjectId", false, col => col.Int32(array => array.AppendRange(_colSubjectId)))
+ .Append("ActivityLabel", false, col => col.String(array => array.AppendRange(_colActivityLabel)))
+ .Append("Timestamp", false, col => col.Int64(array => array.AppendRange(_colTimestamp)))
+ .Append("XAxis", false, col => col.Double(array => array.AppendRange(_colXAxis)))
+ .Append("YAxis", false, col => col.Double(array => array.AppendRange(_colYAxis)))
+ .Append("ZAxis", false, col => col.Double(array => array.AppendRange(_colZAxis)))
+ .Build();
+
+ _recordBatches.Add(recordBatch);
+
+ _colSubjectId.Clear();
+ _colActivityLabel.Clear();
+ _colTimestamp.Clear();
+ _colXAxis.Clear();
+ _colYAxis.Clear();
+ _colZAxis.Clear();
+ }
+ }
+ }
+
+ public async Task<bool> PersistData()
+ {
+ string time = DateTime.Now.ToString("yyyyMMdd_HHmmss");
Review comment:
```suggestion
```
This variable doesn't appear to be used.
##########
File path: csharp/examples/IoTExample/Model/SampleDataset.cs
##########
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+
+namespace IoTExample
+{
+ public class SampleDataset
+ {
+ private int _size;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly BlockingCollection<SensorData> _rows;
+
+ private readonly List<int> _colSubjectId;
+ private readonly List<string> _colActivityLabel;
+ private readonly List<long> _colTimestamp;
+ private readonly List<double> _colXAxis;
+ private readonly List<double> _colYAxis;
+ private readonly List<double> _colZAxis;
+
+ private readonly int _threshold = 1_000_000;
+ private readonly List<RecordBatch> _recordBatches;
+ private readonly MemoryAllocator _memoryAllocator;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataset(int inputs, int capacity)
+ {
+ _inputs = inputs;
+ _capacity = capacity;
+ _rows = new BlockingCollection<SensorData>(_capacity);
+
+ _colSubjectId = new List<int>(capacity);
+ _colActivityLabel = new List<string>();
+ _colTimestamp = new List<long>();
+ _colXAxis = new List<double>();
+ _colYAxis = new List<double>();
+ _colZAxis = new List<double>();
+
+ _recordBatches = new List<RecordBatch>();
+ _memoryAllocator = new NativeMemoryAllocator(alignment: 64);
+ }
+
+ public void Produce()
+ {
+ Random rand = new Random();
+ bool success;
+
+ List<string> keyList = new List<string>(activityLabel.Keys);
+ int count = keyList.Count;
+
+ DateTime now = DateTime.Now;
+ long unixTime = ((DateTimeOffset)now).ToUnixTimeSeconds();
+
+ while (_size < _inputs)
+ {
+ string randomKey = keyList[rand.Next(count)];
+ string label = activityLabel[randomKey];
+
+ // generate missing values
+ if (rand.Next(10_000) == 9_999)
+ {
+ label = null;
+ }
+
+ success = _rows.TryAdd(new SensorData
+ {
+ subjectId = rand.Next(1000, 2001),
+ activityLabel = label,
+ timestamp = unixTime++,
+ x_Axis = rand.NextDouble(),
+ y_Axis = rand.NextDouble(),
+ z_Axis = rand.NextDouble(),
+ });
+
+ if (success)
+ {
+ //Console.WriteLine($"Enqueue Task 0");
+ _size++;
+ }
+ else
+ {
+ Console.WriteLine("Producing is blocked, percent completed is: {0}%", Math.Round((double)_size / _inputs, 4) * 100);
+ }
+ }
+
+ _rows.CompleteAdding();
+ }
+
+ public async void Consume()
Review comment:
This method doesn't need to be `async` since it doesn't `await` anything.
Also - avoid `async void` methods in general. Always return `async Task`, if you can.
##########
File path: csharp/examples/IoTExample/Model/SampleDataset.cs
##########
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+
+namespace IoTExample
+{
+ public class SampleDataset
+ {
+ private int _size;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly BlockingCollection<SensorData> _rows;
Review comment:
Maybe since we are using `async` methods, can we change this to use [Channels](https://devblogs.microsoft.com/dotnet/an-introduction-to-system-threading-channels/) instead? They are very similar, but Channels was made with `async` in mind.
##########
File path: csharp/examples/IoTExample/Model/SampleDataset.cs
##########
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+
+namespace IoTExample
+{
+ public class SampleDataset
+ {
+ private int _size;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly BlockingCollection<SensorData> _rows;
+
+ private readonly List<int> _colSubjectId;
+ private readonly List<string> _colActivityLabel;
+ private readonly List<long> _colTimestamp;
+ private readonly List<double> _colXAxis;
+ private readonly List<double> _colYAxis;
+ private readonly List<double> _colZAxis;
+
+ private readonly int _threshold = 1_000_000;
+ private readonly List<RecordBatch> _recordBatches;
+ private readonly MemoryAllocator _memoryAllocator;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataset(int inputs, int capacity)
+ {
+ _inputs = inputs;
+ _capacity = capacity;
+ _rows = new BlockingCollection<SensorData>(_capacity);
+
+ _colSubjectId = new List<int>(capacity);
+ _colActivityLabel = new List<string>();
+ _colTimestamp = new List<long>();
+ _colXAxis = new List<double>();
+ _colYAxis = new List<double>();
+ _colZAxis = new List<double>();
+
+ _recordBatches = new List<RecordBatch>();
+ _memoryAllocator = new NativeMemoryAllocator(alignment: 64);
+ }
+
+ public void Produce()
+ {
+ Random rand = new Random();
+ bool success;
+
+ List<string> keyList = new List<string>(activityLabel.Keys);
+ int count = keyList.Count;
+
+ DateTime now = DateTime.Now;
+ long unixTime = ((DateTimeOffset)now).ToUnixTimeSeconds();
+
+ while (_size < _inputs)
+ {
+ string randomKey = keyList[rand.Next(count)];
+ string label = activityLabel[randomKey];
+
+ // generate missing values
+ if (rand.Next(10_000) == 9_999)
+ {
+ label = null;
+ }
+
+ success = _rows.TryAdd(new SensorData
+ {
+ subjectId = rand.Next(1000, 2001),
+ activityLabel = label,
+ timestamp = unixTime++,
+ x_Axis = rand.NextDouble(),
+ y_Axis = rand.NextDouble(),
+ z_Axis = rand.NextDouble(),
+ });
+
+ if (success)
+ {
+ //Console.WriteLine($"Enqueue Task 0");
Review comment:
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] eerhardt commented on pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
eerhardt commented on pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#issuecomment-1004434501
@zixi-bwang - some bad merge must have happened. This PR is now affecting A LOT of files. Can you fix it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#issuecomment-1004527023
> @zixi-bwang - some bad merge must have happened. This PR is now affecting A LOT of files. Can you fix it?
Not sure how to fix it, I'll have to close the PR then.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r776843933
##########
File path: csharp/examples/IoTExample/Model/SampleDataPipeline.cs
##########
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+using System.Threading.Channels;
+using System.Threading;
+using System.Collections.Concurrent;
+
+namespace IoTPipelineExample
+{
+ public class SampleDataPipeline
+ {
+ private int _size;
+ private int _partitions;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly Channel<SensorData> _channel;
+ ChannelWriter<SensorData> _writer;
+ ChannelReader<SensorData> _reader;
+
+ private readonly List<ConcurrentBag<int>> _colSubjectIdArrays;
+ private readonly List<ConcurrentBag<string>> _colActivityLabelArrays;
+ private readonly List<ConcurrentBag<long>> _colTimestampArrays;
+ private readonly List<ConcurrentBag<double>> _colXAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colYAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colZAxisArrays;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataPipeline(int concurrencyLevel, int totalSensorData, int queueCapacity)
+ {
+ _partitions = concurrencyLevel;
+ _inputs = totalSensorData;
+ _capacity = queueCapacity;
+ _channel = Channel.CreateBounded<SensorData>(_capacity);
+ _writer = _channel.Writer;
+ _reader = _channel.Reader;
+
+ _colSubjectIdArrays = new List<ConcurrentBag<int>>();
+ _colActivityLabelArrays = new List<ConcurrentBag<string>>();
+ _colTimestampArrays = new List<ConcurrentBag<long>>();
+ _colXAxisArrays = new List<ConcurrentBag<double>>();
+ _colYAxisArrays = new List<ConcurrentBag<double>>();
+ _colZAxisArrays = new List<ConcurrentBag<double>>();
+
+ for (int i = 0; i < _partitions; i++)
+ {
+ _colSubjectIdArrays.Add(new ConcurrentBag<int>());
Review comment:
I have refactored the code per your suggestion to use multiple writers and 1 reader only, and writes data directly to builder.
And I also modified the example to emulate below scenario:
- There are around 9,000 sensors(subjects) in total
- Each sensor can send hundreds or thousands events during a short time of period
- When reading from channel, data is appended to each array builder hashed by SubjectId as the key of builderId
- When building RecordBatch, the array builders with the same builderId are built together
- The final Arrow file contains 9,000 RecordBatches
Do you think this a good scenario for real world applications in terms of utilizing Arrow RecordBatch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #11874: A Real-time C# memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#issuecomment-987128831
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
Thanks for opening a pull request!
If this is not a [minor PR](https://github.com/apache/arrow/blob/master/CONTRIBUTING.md#Minor-Fixes). Could you open an issue for this pull request on JIRA? https://issues.apache.org/jira/browse/ARROW
Opening JIRAs ahead of time contributes to the [Openness](http://theapacheway.com/open/#:~:text=Openness%20allows%20new%20users%20the,must%20happen%20in%20the%20open.) of the Apache Arrow project.
Then could you also rename pull request title in the following format?
ARROW-${JIRA_ID}: [${COMPONENT}] ${SUMMARY}
or
MINOR: [${COMPONENT}] ${SUMMARY}
See also:
* [Other pull requests](https://github.com/apache/arrow/pulls/)
* [Contribution Guidelines - How to contribute patches](https://arrow.apache.org/docs/developers/contributing.html#how-to-contribute-patches)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r772884571
##########
File path: csharp/examples/IoTExample/Model/SampleDataset.cs
##########
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+
+namespace IoTExample
+{
+ public class SampleDataset
+ {
+ private int _size;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly BlockingCollection<SensorData> _rows;
Review comment:
The code has been refactored, please review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r776838685
##########
File path: csharp/examples/IoTExample/Model/SampleDataPipeline.cs
##########
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+using System.Threading.Channels;
+using System.Threading;
+using System.Collections.Concurrent;
+
+namespace IoTPipelineExample
+{
+ public class SampleDataPipeline
+ {
+ private int _size;
+ private int _partitions;
+ private readonly int _inputs;
+ private readonly int _capacity;
Review comment:
Not really, removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] kou commented on pull request #11874: MINOR: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
kou commented on pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#issuecomment-987442947
I think that this is not a MINOR change. See our MINOR definition: https://github.com/apache/arrow/blob/master/CONTRIBUTING.md#Minor-Fixes
> Any functionality change should have a JIRA opened. For minor changes that affect documentation, you do not need to open up a JIRA. Instead you can prefix the title of your PR with "MINOR: " if meets the following guidelines:
>
> * Grammar, usage and spelling fixes that affect no more than 2 files
> * Documentation updates affecting no more than 2 files and not more than 500 words.
Could you open an issue on JIRA? https://issues.apache.org/jira/browse/ARROW
See also the auto-generated comment: https://github.com/apache/arrow/pull/11874#issuecomment-987128831
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] eerhardt commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
eerhardt commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r773296758
##########
File path: csharp/examples/IoTExample/Model/SampleDataPipeline.cs
##########
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+using System.Threading.Channels;
+using System.Threading;
+using System.Collections.Concurrent;
+
+namespace IoTPipelineExample
+{
+ public class SampleDataPipeline
+ {
+ private int _size;
+ private int _partitions;
+ private readonly int _inputs;
+ private readonly int _capacity;
Review comment:
Is this field necessary?
##########
File path: csharp/examples/IoTExample/Model/SampleDataPipeline.cs
##########
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+using System.Threading.Channels;
+using System.Threading;
+using System.Collections.Concurrent;
+
+namespace IoTPipelineExample
+{
+ public class SampleDataPipeline
+ {
+ private int _size;
+ private int _partitions;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly Channel<SensorData> _channel;
+ ChannelWriter<SensorData> _writer;
+ ChannelReader<SensorData> _reader;
+
+ private readonly List<ConcurrentBag<int>> _colSubjectIdArrays;
+ private readonly List<ConcurrentBag<string>> _colActivityLabelArrays;
+ private readonly List<ConcurrentBag<long>> _colTimestampArrays;
+ private readonly List<ConcurrentBag<double>> _colXAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colYAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colZAxisArrays;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataPipeline(int concurrencyLevel, int totalSensorData, int queueCapacity)
+ {
+ _partitions = concurrencyLevel;
+ _inputs = totalSensorData;
Review comment:
I find it confusing to use two different names here. Can we pick one and use it both for the field and the parameter?
##########
File path: csharp/examples/IoTExample/Model/SampleDataPipeline.cs
##########
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+using System.Threading.Channels;
+using System.Threading;
+using System.Collections.Concurrent;
+
+namespace IoTPipelineExample
+{
+ public class SampleDataPipeline
+ {
+ private int _size;
+ private int _partitions;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly Channel<SensorData> _channel;
+ ChannelWriter<SensorData> _writer;
+ ChannelReader<SensorData> _reader;
+
+ private readonly List<ConcurrentBag<int>> _colSubjectIdArrays;
+ private readonly List<ConcurrentBag<string>> _colActivityLabelArrays;
+ private readonly List<ConcurrentBag<long>> _colTimestampArrays;
+ private readonly List<ConcurrentBag<double>> _colXAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colYAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colZAxisArrays;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataPipeline(int concurrencyLevel, int totalSensorData, int queueCapacity)
+ {
+ _partitions = concurrencyLevel;
+ _inputs = totalSensorData;
+ _capacity = queueCapacity;
+ _channel = Channel.CreateBounded<SensorData>(_capacity);
+ _writer = _channel.Writer;
+ _reader = _channel.Reader;
+
+ _colSubjectIdArrays = new List<ConcurrentBag<int>>();
+ _colActivityLabelArrays = new List<ConcurrentBag<string>>();
+ _colTimestampArrays = new List<ConcurrentBag<long>>();
+ _colXAxisArrays = new List<ConcurrentBag<double>>();
+ _colYAxisArrays = new List<ConcurrentBag<double>>();
+ _colZAxisArrays = new List<ConcurrentBag<double>>();
+
+ for (int i = 0; i < _partitions; i++)
+ {
+ _colSubjectIdArrays.Add(new ConcurrentBag<int>());
Review comment:
ConcurrentBag is unordered. Meaning when you enumerate it, you can get the results in a different order than what they went in. Is that intended?
https://docs.microsoft.com/en-us/dotnet/api/system.collections.concurrent.concurrentbag-1?view=net-6.0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r776838967
##########
File path: csharp/examples/IoTExample/Model/SampleDataset.cs
##########
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+
+namespace IoTExample
+{
+ public class SampleDataset
+ {
+ private int _size;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly BlockingCollection<SensorData> _rows;
+
+ private readonly List<int> _colSubjectId;
+ private readonly List<string> _colActivityLabel;
+ private readonly List<long> _colTimestamp;
+ private readonly List<double> _colXAxis;
+ private readonly List<double> _colYAxis;
+ private readonly List<double> _colZAxis;
+
+ private readonly int _threshold = 1_000_000;
+ private readonly List<RecordBatch> _recordBatches;
+ private readonly MemoryAllocator _memoryAllocator;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataset(int inputs, int capacity)
+ {
+ _inputs = inputs;
+ _capacity = capacity;
+ _rows = new BlockingCollection<SensorData>(_capacity);
+
+ _colSubjectId = new List<int>(capacity);
+ _colActivityLabel = new List<string>();
+ _colTimestamp = new List<long>();
+ _colXAxis = new List<double>();
+ _colYAxis = new List<double>();
+ _colZAxis = new List<double>();
+
+ _recordBatches = new List<RecordBatch>();
+ _memoryAllocator = new NativeMemoryAllocator(alignment: 64);
+ }
+
+ public void Produce()
+ {
+ Random rand = new Random();
+ bool success;
+
+ List<string> keyList = new List<string>(activityLabel.Keys);
+ int count = keyList.Count;
+
+ DateTime now = DateTime.Now;
+ long unixTime = ((DateTimeOffset)now).ToUnixTimeSeconds();
+
+ while (_size < _inputs)
+ {
+ string randomKey = keyList[rand.Next(count)];
+ string label = activityLabel[randomKey];
+
+ // generate missing values
+ if (rand.Next(10_000) == 9_999)
+ {
+ label = null;
+ }
+
+ success = _rows.TryAdd(new SensorData
+ {
+ subjectId = rand.Next(1000, 2001),
+ activityLabel = label,
+ timestamp = unixTime++,
+ x_Axis = rand.NextDouble(),
+ y_Axis = rand.NextDouble(),
+ z_Axis = rand.NextDouble(),
+ });
+
+ if (success)
+ {
+ //Console.WriteLine($"Enqueue Task 0");
+ _size++;
+ }
+ else
+ {
+ Console.WriteLine("Producing is blocked, percent completed is: {0}%", Math.Round((double)_size / _inputs, 4) * 100);
+ }
+ }
+
+ _rows.CompleteAdding();
+ }
+
+ public async void Consume()
Review comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#issuecomment-989303855
> One quick question/suggestion: Can you add a small `README` to the example explaining what it is showing? Describe the scenario, etc.
Hi @eerhardt, I added the readme also updated the example, please review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r776838749
##########
File path: csharp/examples/IoTExample/Model/SampleDataPipeline.cs
##########
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+using System.Threading.Channels;
+using System.Threading;
+using System.Collections.Concurrent;
+
+namespace IoTPipelineExample
+{
+ public class SampleDataPipeline
+ {
+ private int _size;
+ private int _partitions;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly Channel<SensorData> _channel;
+ ChannelWriter<SensorData> _writer;
+ ChannelReader<SensorData> _reader;
+
+ private readonly List<ConcurrentBag<int>> _colSubjectIdArrays;
+ private readonly List<ConcurrentBag<string>> _colActivityLabelArrays;
+ private readonly List<ConcurrentBag<long>> _colTimestampArrays;
+ private readonly List<ConcurrentBag<double>> _colXAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colYAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colZAxisArrays;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataPipeline(int concurrencyLevel, int totalSensorData, int queueCapacity)
+ {
+ _partitions = concurrencyLevel;
+ _inputs = totalSensorData;
Review comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r773548623
##########
File path: csharp/examples/IoTExample/Model/SampleDataPipeline.cs
##########
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+using System.Threading.Channels;
+using System.Threading;
+using System.Collections.Concurrent;
+
+namespace IoTPipelineExample
+{
+ public class SampleDataPipeline
+ {
+ private int _size;
+ private int _partitions;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly Channel<SensorData> _channel;
+ ChannelWriter<SensorData> _writer;
+ ChannelReader<SensorData> _reader;
+
+ private readonly List<ConcurrentBag<int>> _colSubjectIdArrays;
+ private readonly List<ConcurrentBag<string>> _colActivityLabelArrays;
+ private readonly List<ConcurrentBag<long>> _colTimestampArrays;
+ private readonly List<ConcurrentBag<double>> _colXAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colYAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colZAxisArrays;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataPipeline(int concurrencyLevel, int totalSensorData, int queueCapacity)
+ {
+ _partitions = concurrencyLevel;
+ _inputs = totalSensorData;
+ _capacity = queueCapacity;
+ _channel = Channel.CreateBounded<SensorData>(_capacity);
+ _writer = _channel.Writer;
+ _reader = _channel.Reader;
+
+ _colSubjectIdArrays = new List<ConcurrentBag<int>>();
+ _colActivityLabelArrays = new List<ConcurrentBag<string>>();
+ _colTimestampArrays = new List<ConcurrentBag<long>>();
+ _colXAxisArrays = new List<ConcurrentBag<double>>();
+ _colYAxisArrays = new List<ConcurrentBag<double>>();
+ _colZAxisArrays = new List<ConcurrentBag<double>>();
+
+ for (int i = 0; i < _partitions; i++)
+ {
+ _colSubjectIdArrays.Add(new ConcurrentBag<int>());
Review comment:
My intention was to find a data structure that has below characteristics:
1. Inserting by row
2. Querying by column
3. The "results in a different order than what they went in" does not matter, but the kth item in one column should belong to the same row as the kth item in another column
4. So all of the column arrays can be appended to a RecordBatch
It seems DataTable is a good candidate, or could you recommend other light weight data structures instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on pull request #11874: MINOR: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#issuecomment-987140433
Hi, @eerhardt
I added a usage example, could you review?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r772068587
##########
File path: csharp/examples/IoTExample/Model/SampleDataset.cs
##########
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+
+namespace IoTExample
+{
+ public class SampleDataset
+ {
+ private int _size;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly BlockingCollection<SensorData> _rows;
Review comment:
Sure, I'll try to use Channel.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] eerhardt commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
eerhardt commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r776000123
##########
File path: csharp/examples/IoTExample/Model/SampleDataPipeline.cs
##########
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+using System.Threading.Channels;
+using System.Threading;
+using System.Collections.Concurrent;
+
+namespace IoTPipelineExample
+{
+ public class SampleDataPipeline
+ {
+ private int _size;
+ private int _partitions;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly Channel<SensorData> _channel;
+ ChannelWriter<SensorData> _writer;
+ ChannelReader<SensorData> _reader;
+
+ private readonly List<ConcurrentBag<int>> _colSubjectIdArrays;
+ private readonly List<ConcurrentBag<string>> _colActivityLabelArrays;
+ private readonly List<ConcurrentBag<long>> _colTimestampArrays;
+ private readonly List<ConcurrentBag<double>> _colXAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colYAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colZAxisArrays;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataPipeline(int concurrencyLevel, int totalSensorData, int queueCapacity)
+ {
+ _partitions = concurrencyLevel;
+ _inputs = totalSensorData;
+ _capacity = queueCapacity;
+ _channel = Channel.CreateBounded<SensorData>(_capacity);
+ _writer = _channel.Writer;
+ _reader = _channel.Reader;
+
+ _colSubjectIdArrays = new List<ConcurrentBag<int>>();
+ _colActivityLabelArrays = new List<ConcurrentBag<string>>();
+ _colTimestampArrays = new List<ConcurrentBag<long>>();
+ _colXAxisArrays = new List<ConcurrentBag<double>>();
+ _colYAxisArrays = new List<ConcurrentBag<double>>();
+ _colZAxisArrays = new List<ConcurrentBag<double>>();
+
+ for (int i = 0; i < _partitions; i++)
+ {
+ _colSubjectIdArrays.Add(new ConcurrentBag<int>());
Review comment:
> but the kth item in one column should belong to the same row as the kth item in another column
Since a "bag" is unordered, there is no such thing as the "kth item". Notice there is no indexer. You can enumerate the items, but you aren't guaranteed the order they are returned.
> It seems DataTable is a good candidate, or could you recommend other light weight data structures instead?
I wouldn't use DataTable. Instead, I would eliminate the concurrency in `ReadFromChannel`. Have many "writer" threads all writing items to the channel. But then only have 1 "reader" thread that pulls items off the channel and writes them directly to a `RecordBatch.Builder`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#issuecomment-989303372
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang closed pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang closed pull request #11874:
URL: https://github.com/apache/arrow/pull/11874
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r776838863
##########
File path: csharp/examples/IoTExample/Model/SampleDataset.cs
##########
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Diagnostics;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+
+namespace IoTExample
+{
+ public class SampleDataset
+ {
+ private int _size;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly BlockingCollection<SensorData> _rows;
+
+ private readonly List<int> _colSubjectId;
+ private readonly List<string> _colActivityLabel;
+ private readonly List<long> _colTimestamp;
+ private readonly List<double> _colXAxis;
+ private readonly List<double> _colYAxis;
+ private readonly List<double> _colZAxis;
+
+ private readonly int _threshold = 1_000_000;
+ private readonly List<RecordBatch> _recordBatches;
+ private readonly MemoryAllocator _memoryAllocator;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataset(int inputs, int capacity)
+ {
+ _inputs = inputs;
+ _capacity = capacity;
+ _rows = new BlockingCollection<SensorData>(_capacity);
+
+ _colSubjectId = new List<int>(capacity);
+ _colActivityLabel = new List<string>();
+ _colTimestamp = new List<long>();
+ _colXAxis = new List<double>();
+ _colYAxis = new List<double>();
+ _colZAxis = new List<double>();
+
+ _recordBatches = new List<RecordBatch>();
+ _memoryAllocator = new NativeMemoryAllocator(alignment: 64);
+ }
+
+ public void Produce()
+ {
+ Random rand = new Random();
+ bool success;
+
+ List<string> keyList = new List<string>(activityLabel.Keys);
+ int count = keyList.Count;
+
+ DateTime now = DateTime.Now;
+ long unixTime = ((DateTimeOffset)now).ToUnixTimeSeconds();
+
+ while (_size < _inputs)
+ {
+ string randomKey = keyList[rand.Next(count)];
+ string label = activityLabel[randomKey];
+
+ // generate missing values
+ if (rand.Next(10_000) == 9_999)
+ {
+ label = null;
+ }
+
+ success = _rows.TryAdd(new SensorData
+ {
+ subjectId = rand.Next(1000, 2001),
+ activityLabel = label,
+ timestamp = unixTime++,
+ x_Axis = rand.NextDouble(),
+ y_Axis = rand.NextDouble(),
+ z_Axis = rand.NextDouble(),
+ });
+
+ if (success)
+ {
+ //Console.WriteLine($"Enqueue Task 0");
+ _size++;
+ }
+ else
+ {
+ Console.WriteLine("Producing is blocked, percent completed is: {0}%", Math.Round((double)_size / _inputs, 4) * 100);
+ }
+ }
+
+ _rows.CompleteAdding();
+ }
+
+ public async void Consume()
+ {
+ while (!_rows.IsCompleted)
+ {
+ if (!_rows.TryTake(out SensorData item, 3000))
+ {
+ Console.WriteLine("Consuming is blocked!");
+ }
+ else
+ {
+ //Console.WriteLine($"Dequeue Task 1");
+ if (item != null && item.subjectId != null)
+ {
+ _colSubjectId.Add((int)item.subjectId);
+ _colActivityLabel.Add(item.activityLabel);
+ _colTimestamp.Add((long)item.timestamp);
+ _colXAxis.Add((double)item.x_Axis);
+ _colYAxis.Add((double)item.y_Axis);
+ _colZAxis.Add((double)item.z_Axis);
+ }
+ }
+
+ if (_colSubjectId.Count == _threshold)
+ {
+ // Build a record batch using the Fluent API
+ var recordBatch = new RecordBatch.Builder(_memoryAllocator)
+ .Append("SubjectId", false, col => col.Int32(array => array.AppendRange(_colSubjectId)))
+ .Append("ActivityLabel", false, col => col.String(array => array.AppendRange(_colActivityLabel)))
+ .Append("Timestamp", false, col => col.Int64(array => array.AppendRange(_colTimestamp)))
+ .Append("XAxis", false, col => col.Double(array => array.AppendRange(_colXAxis)))
+ .Append("YAxis", false, col => col.Double(array => array.AppendRange(_colYAxis)))
+ .Append("ZAxis", false, col => col.Double(array => array.AppendRange(_colZAxis)))
+ .Build();
+
+ _recordBatches.Add(recordBatch);
+
+ _colSubjectId.Clear();
+ _colActivityLabel.Clear();
+ _colTimestamp.Clear();
+ _colXAxis.Clear();
+ _colYAxis.Clear();
+ _colZAxis.Clear();
+ }
+ }
+ }
+
+ public async Task<bool> PersistData()
+ {
+ string time = DateTime.Now.ToString("yyyyMMdd_HHmmss");
+ Stopwatch stopwatch = new Stopwatch();
+ stopwatch.Start();
+
+ Schema schema = _recordBatches[0].Schema;
+ // Write record batch to a file
+ using (var stream = File.OpenWrite(@"c:\temp\data\iotbigdata.arrow"))
Review comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r776843933
##########
File path: csharp/examples/IoTExample/Model/SampleDataPipeline.cs
##########
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+using System.Threading.Channels;
+using System.Threading;
+using System.Collections.Concurrent;
+
+namespace IoTPipelineExample
+{
+ public class SampleDataPipeline
+ {
+ private int _size;
+ private int _partitions;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly Channel<SensorData> _channel;
+ ChannelWriter<SensorData> _writer;
+ ChannelReader<SensorData> _reader;
+
+ private readonly List<ConcurrentBag<int>> _colSubjectIdArrays;
+ private readonly List<ConcurrentBag<string>> _colActivityLabelArrays;
+ private readonly List<ConcurrentBag<long>> _colTimestampArrays;
+ private readonly List<ConcurrentBag<double>> _colXAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colYAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colZAxisArrays;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataPipeline(int concurrencyLevel, int totalSensorData, int queueCapacity)
+ {
+ _partitions = concurrencyLevel;
+ _inputs = totalSensorData;
+ _capacity = queueCapacity;
+ _channel = Channel.CreateBounded<SensorData>(_capacity);
+ _writer = _channel.Writer;
+ _reader = _channel.Reader;
+
+ _colSubjectIdArrays = new List<ConcurrentBag<int>>();
+ _colActivityLabelArrays = new List<ConcurrentBag<string>>();
+ _colTimestampArrays = new List<ConcurrentBag<long>>();
+ _colXAxisArrays = new List<ConcurrentBag<double>>();
+ _colYAxisArrays = new List<ConcurrentBag<double>>();
+ _colZAxisArrays = new List<ConcurrentBag<double>>();
+
+ for (int i = 0; i < _partitions; i++)
+ {
+ _colSubjectIdArrays.Add(new ConcurrentBag<int>());
Review comment:
I have refactored the code per your suggestion to use multiple writers and 1 reader only, and writes data directly to builder.
And I also modified the example to emulate below scenario:
- There are around 9,000 sensors(subjects) in total
- Each sensor can send hundreds or thousands events during a short time of period
- When reading from channel, data is appended to each array builder hashed by SubjectId as the key of builderId
- When building RecordBatch, the array builders with the same builderId are built together
- The final Arrow file contains 9,000 RecordBatches
Do you think this is a good scenario for real world use cases in terms of utilizing Arrow RecordBatch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] zixi-bwang commented on a change in pull request #11874: ARROW-15037: [C#] A Real-time memory-based data analytics example
Posted by GitBox <gi...@apache.org>.
zixi-bwang commented on a change in pull request #11874:
URL: https://github.com/apache/arrow/pull/11874#discussion_r776843933
##########
File path: csharp/examples/IoTExample/Model/SampleDataPipeline.cs
##########
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using System.Collections.Generic;
+using Apache.Arrow;
+using Apache.Arrow.Ipc;
+using Apache.Arrow.Memory;
+using System.Threading.Channels;
+using System.Threading;
+using System.Collections.Concurrent;
+
+namespace IoTPipelineExample
+{
+ public class SampleDataPipeline
+ {
+ private int _size;
+ private int _partitions;
+ private readonly int _inputs;
+ private readonly int _capacity;
+ private readonly Channel<SensorData> _channel;
+ ChannelWriter<SensorData> _writer;
+ ChannelReader<SensorData> _reader;
+
+ private readonly List<ConcurrentBag<int>> _colSubjectIdArrays;
+ private readonly List<ConcurrentBag<string>> _colActivityLabelArrays;
+ private readonly List<ConcurrentBag<long>> _colTimestampArrays;
+ private readonly List<ConcurrentBag<double>> _colXAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colYAxisArrays;
+ private readonly List<ConcurrentBag<double>> _colZAxisArrays;
+
+ public Dictionary<string, string> activityLabel = new Dictionary<string, string>()
+ {
+ {"walking", "A"},
+ {"jogging", "B"},
+ {"stairs", "C"},
+ {"sitting", "D"},
+ {"standing", "E"},
+ {"typing", "F"},
+ {"teeth", "G"},
+ {"soup", "H"},
+ {"chips", "I"},
+ {"pasta", "J"},
+ {"drinking", "K"},
+ {"sandwich", "L"},
+ {"kicking", "M"},
+ {"catch", "O"},
+ {"dribbling", "P"},
+ {"writing", "Q"},
+ {"clapping", "R"},
+ {"folding", "S"},
+ };
+
+ public SampleDataPipeline(int concurrencyLevel, int totalSensorData, int queueCapacity)
+ {
+ _partitions = concurrencyLevel;
+ _inputs = totalSensorData;
+ _capacity = queueCapacity;
+ _channel = Channel.CreateBounded<SensorData>(_capacity);
+ _writer = _channel.Writer;
+ _reader = _channel.Reader;
+
+ _colSubjectIdArrays = new List<ConcurrentBag<int>>();
+ _colActivityLabelArrays = new List<ConcurrentBag<string>>();
+ _colTimestampArrays = new List<ConcurrentBag<long>>();
+ _colXAxisArrays = new List<ConcurrentBag<double>>();
+ _colYAxisArrays = new List<ConcurrentBag<double>>();
+ _colZAxisArrays = new List<ConcurrentBag<double>>();
+
+ for (int i = 0; i < _partitions; i++)
+ {
+ _colSubjectIdArrays.Add(new ConcurrentBag<int>());
Review comment:
I have refactored the code per your suggestion to use multiple writers and 1 reader only, and writes data directly to builder.
And I also modified the example to emulate below scenario:
- There are around 9,000 sensors(subjects) in total
- Each sensor can send hundreds or thousands events during a short time of period
- When reading from channel, data is appended to each array builder hashed by SubjectId as the key of builderId
- When building RecordBatch, the array builders with the same builderId are built together
- The final Arrow file contains 9,000 RecordBatches
Do you think this is a good scenario for real world use cases in terms of utilizing Arrow file and RecordBatch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org