# Streaming online random forests for anomly detection
# Now adding Apache Beam for the online random forest generator making sure that we the best infrastructure for maximizing the scaling of the online anomaly detectionfrom sklearn.ensemble import RandomForestClassifier
import numpy as np
class OnlineRandomForest:
def __init__(self, n_estimators=100, max_depth=None, max_features='auto'):
self.n_estimators = n_estimators
self.max_depth = max_depth
self.max_features = max_features
self.forest = []
def update(self, X, y):
# Convert the label to a numpy array
y = np.array([y])
# Create a new tree and add it to the forest
tree = RandomForestClassifier(
n_estimators=1,
max_depth=self.max_depth,
max_features=self.max_features
)
tree.fit(X, y)
self.forest.append(tree)
# Remove the oldest tree if the forest exceeds the desired number of trees
if len(self.forest) > self.n_estimators:
self.forest.pop(0)
def predict(self, X, anomaly_threshold):
# Make predictions by averaging the predictions from all trees in the forest
predictions = np.zeros((X.shape[0], len(self.forest)))
for i, tree in enumerate(self.forest):
predictions[:, i] = tree.predict(X)
# Calculate the average prediction for each sample
average_predictions = np.mean(predictions, axis=1)
# Identify anomalies based on the anomaly threshold
anomalies = average_predictions > anomaly_threshold
return average_predictions, anomalies
return np.mean(predictions, axis=1)
# Simulated data generator
def data_generator():
while True:
# Generate random features and labels
features = np.random.rand(1, 5) # Replace with your actual feature generation logic
label = np.random.randint(2) # Replace with your actual label generation logic
yield features, label
# Example usage
# Initialize the online random forest
online_rf = OnlineRandomForest(n_estimators=10, max_depth=5, max_features='sqrt')
# Create a generator object for streaming data
data_stream = data_generator()
prediction_array = []
anomalies_array = []
# Continuously update the forest as new examples arrive
for _ in range(100): # Replace 100 with the desired number of iterations
# Get the next example from the data stream
features, label = next(data_stream)
# Update the online random forest with the new example
online_rf.update(features, label)
# Make predictions for new examples
new_example = np.random.rand(1, 5) # Replace with your new example's features
prediction = online_rf.predict(new_example,anomaly_threshold=0.65)
prediction_array.append(prediction[0])
if prediction[1]== True:
anomalies_array.append(prediction[0])
print("Prediction:", prediction)Prediction: (array([0.]), array([False]))
Prediction: (array([0.]), array([False]))
Prediction: (array([0.]), array([False]))
Prediction: (array([0.]), array([False]))
Prediction: (array([0.]), array([False]))
Prediction: (array([0.16666667]), array([False]))
Prediction: (array([0.14285714]), array([False]))
Prediction: (array([0.25]), array([False]))
Prediction: (array([0.22222222]), array([False]))
Prediction: (array([0.2]), array([False]))
Prediction: (array([0.3]), array([False]))
Prediction: (array([0.3]), array([False]))
Prediction: (array([0.3]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.7]), array([ True]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.7]), array([ True]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.3]), array([False]))
Prediction: (array([0.3]), array([False]))
Prediction: (array([0.3]), array([False]))
Prediction: (array([0.3]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.7]), array([ True]))
Prediction: (array([0.7]), array([ True]))
Prediction: (array([0.7]), array([ True]))
Prediction: (array([0.8]), array([ True]))
Prediction: (array([0.8]), array([ True]))
Prediction: (array([0.7]), array([ True]))
Prediction: (array([0.7]), array([ True]))
Prediction: (array([0.7]), array([ True]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.6]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.5]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.4]), array([False]))
Prediction: (array([0.3]), array([False]))
Prediction: (array([0.3]), array([False]))
Prediction: (array([0.2]), array([False]))
Prediction: (array([0.1]), array([False]))
Prediction: (array([0.]), array([False]))
Prediction: (array([0.1]), array([False]))
Prediction: (array([0.1]), array([False]))
Prediction: (array([0.2]), array([False]))
Prediction: (array([0.2]), array([False]))
Prediction: (array([0.2]), array([False]))
Prediction: (array([0.2]), array([False]))
Prediction: (array([0.2]), array([False]))
Prediction: (array([0.2]), array([False]))
Prediction: (array([0.3]), array([False]))
Prediction: (array([0.3]), array([False]))
import matplotlib.pyplot as plt
# Assuming you have the 'average_predictions' array available
# Create a histogram of prediction values
plt.hist(prediction_array, bins=20)
# Plot settings
plt.xlabel('Prediction Value')
plt.ylabel('Frequency')
plt.title('Histogram of Prediction Values')
# Show the plot
plt.show()
import matplotlib.pyplot as plt
# Assuming you have the 'average_predictions' and 'anomalies' arrays available
# Scatter plot of average predictions
plt.figure(figsize=(10, 5))
plt.subplot(211)
plt.scatter(range(len(prediction_array)), prediction_array, label='Average Predictions')
plt.xlabel('Sample Index')
plt.ylabel('Average Prediction')
plt.title('Average Predictions')
plt.legend()
# Scatter plot of anomalies
plt.subplot(212)
plt.scatter(range(len(anomalies_array)), anomalies_array, c='red', label='Anomalies')
plt.xlabel('Sample Index')
plt.ylabel('Anomaly')
plt.title('Anomalies')
plt.legend()
# Adjust layout and spacing
plt.tight_layout()
# Show the plots
plt.show()
! pip install apache-beamCollecting apache-beam
Downloading apache_beam-2.48.0-cp38-cp38-macosx_10_9_x86_64.whl (5.1 MB)
acosx_10_9_x86_64.whl (473 kB)
ongo<5.0.0,>=3.8.0
Downloading pymongo-4.3.3-cp38-cp38-macosx_10_9_x86_64.whl (381 kB)
ent already satisfied: python-dateutil<3,>=2.8.0 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from apache-beam) (2.8.2)
Collecting orjson<4.0
Downloading orjson-3.9.0-cp38-cp38-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl (240 kB)
od<2.0,>=1.7
Downloading crcmod-1.7.tar.gz (89 kB)
ent already satisfied: protobuf<4.24.0,>=3.20.3 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from apache-beam) (3.20.3)
Requirement already satisfied: dill<0.3.2,>=0.3.1.1 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from apache-beam) (0.3.1.1)
Collecting fastavro<2,>=0.23.6
Downloading fastavro-1.7.4-cp38-cp38-macosx_10_15_x86_64.whl (526 kB)
ent already satisfied: typing-extensions>=3.7.0 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from apache-beam) (4.5.0)
Collecting hdfs<3.0.0,>=2.1.0
Downloading hdfs-2.7.0-py3-none-any.whl (34 kB)
Requirement already satisfied: grpcio!=1.48.0,<2,>=1.33.1 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from apache-beam) (1.49.1)
Requirement already satisfied: requests<3.0.0,>=2.24.0 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from apache-beam) (2.28.2)
Requirement already satisfied: pytz>=2018.3 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from apache-beam) (2023.3)
Requirement already satisfied: numpy<1.25.0,>=1.14.3 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from apache-beam) (1.24.3)
Requirement already satisfied: regex>=2020.6.8 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from apache-beam) (2022.10.31)
Requirement already satisfied: pyarrow<12.0.0,>=3.0.0 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from apache-beam) (10.0.1)
Requirement already satisfied: pyparsing!=3.0.0,!=3.0.1,!=3.0.2,!=3.0.3,<4,>=2.4.2; python_version > "3.0" in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from httplib2<0.23.0,>=0.8->apache-beam) (3.0.9)
Requirement already satisfied: dnspython<3.0.0,>=1.16.0 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from pymongo<5.0.0,>=3.8.0->apache-beam) (2.3.0)
Requirement already satisfied: six>=1.5 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from python-dateutil<3,>=2.8.0->apache-beam) (1.16.0)
Requirement already satisfied: docopt in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from hdfs<3.0.0,>=2.1.0->apache-beam) (0.6.2)
Requirement already satisfied: idna<4,>=2.5 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from requests<3.0.0,>=2.24.0->apache-beam) (3.4)
Requirement already satisfied: urllib3<1.27,>=1.21.1 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from requests<3.0.0,>=2.24.0->apache-beam) (1.26.15)
Requirement already satisfied: certifi>=2017.4.17 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from requests<3.0.0,>=2.24.0->apache-beam) (2022.12.7)
Requirement already satisfied: charset-normalizer<4,>=2 in /Users/sugeerthmurugesan/opt/miniconda3/lib/python3.8/site-packages (from requests<3.0.0,>=2.24.0->apache-beam) (2.1.1)
Building wheels for collected packages: crcmod
Building wheel for crcmod (setup.py) ... od: filename=crcmod-1.7-cp38-cp38-macosx_10_9_x86_64.whl size=22313 sha256=bf80d1fdb12a4bc6ba501d64a6f0a3bbe6fd13ec7bc3d7502b48fcc12f14b489
Stored in directory: /Users/sugeerthmurugesan/Library/Caches/pip/wheels/ca/5a/02/f3acf982a026f3319fb3e798a8dca2d48fafee7761788562e9
Successfully built crcmod
Installing collected packages: zstandard, httplib2, pymongo, fasteners, orjson, objsize, proto-plus, pydot, cloudpickle, crcmod, fastavro, hdfs, apache-beam
Successfully installed apache-beam-2.48.0 cloudpickle-2.2.1 crcmod-1.7 fastavro-1.7.4 fasteners-0.18 hdfs-2.7.0 httplib2-0.22.0 objsize-0.6.1 orjson-3.9.0 proto-plus-1.22.2 pydot-1.4.2 pymongo-4.3.3 zstandard-0.21.0
import apache_beam as beam
from sklearn.ensemble import RandomForestClassifier
import numpy as np
import random
class OnlineRandomForest:
def __init__(self, n_estimators=100, max_depth=None, max_features='auto'):
self.n_estimators = n_estimators
self.max_depth = max_depth
self.max_features = max_features
self.forest = []
def update(self, X, y):
# Convert the label to a numpy array
y = np.array([y])
# Create a new tree and add it to the forest
tree = RandomForestClassifier(
n_estimators=1,
max_depth=self.max_depth,
max_features=self.max_features
)
tree.fit(X, y)
self.forest.append(tree)
# Remove the oldest tree if the forest exceeds the desired number of trees
if len(self.forest) > self.n_estimators:
self.forest.pop(0)
def predict(self, X, anomaly_threshold=0.5):
# Make predictions by averaging the predictions from all trees in the forest
predictions = np.zeros((X.shape[0], len(self.forest)))
for i, tree in enumerate(self.forest):
predictions[:, i] = tree.predict(X)
# Calculate the average prediction for each sample
average_predictions = np.mean(predictions, axis=1)
# Identify anomalies based on the anomaly threshold
anomalies = average_predictions > anomaly_threshold
return average_predictions, anomalies
class AnomalyDetectionDoFn(beam.DoFn):
def __init__(self, online_rf):
self.online_rf = online_rf
def process(self, element):
data_point = element
features = np.array([data_point])
average_predictions, anomalies = self.online_rf.predict(features)
yield (data_point, average_predictions[0], anomalies[0])
def generate_random_data():
while True:
yield random.uniform(0, 1)
def run_anomaly_detection_pipeline():
print("DATASETS")
with beam.Pipeline() as pipeline:
print("DATASETS")
# Generate random data stream
data_stream = pipeline | "Generate Data" >> beam.Create(generate_random_data())
# Create an instance of the online random forest
online_rf = OnlineRandomForest()
# Apply anomaly detection using the AnomalyDetectionDoFn
anomaly_scores = data_stream | "Detect Anomalies" >> beam.ParDo(AnomalyDetectionDoFn(online_rf))
# Calculate the average predictions using CombinePerKey transform
average_predictions = anomaly_scores | "Average Predictions" >> beam.CombinePerKey(np.mean)
# Define a sink to write the average predictions to a file, database, or any other output destination
average_predictions | "Write Average Predictions" >> beam.io.WriteToText("average_predictions.txt")
if __name__ == '__main__':
run_anomaly_detection_pipeline()