Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics and data integration. It's being used even in the biggest financial, telecom and manufacturing organizations in the World. You can build real-time or near-real-time Strategy applications based on it.
Why would you like to stream data from Kafka directly to Strategy? First, you may want to avoid all intermediate steps like databases, ETL processes, staging areas. Simplification of data processing means faster results and less chances for errors. Secondly, you may want to build real-time or near-real-time dashboards to monitor the most vital processes within your organization. Thirdly, you may want to blend real-time data streamed from Kafka with data coming from your other data sources like databases with aggregated historical data or flat files with forecast data.
Strategy Analytics is a perfect platform to put it all together!
Note - I've added color formatting in the code below (http://hilite.me/). Code copied from here might not work due to indentation issues and some problems with quotes (""). In that case, use attached file with sample code.
### Kafka Producer - reading Windows system stats like CPU utilization and publishing it into a topic
### 2023-07-13 / Robert Prochowicz
# tested with Strategy 2021 update 9 and mstrio 11.3.10.101
from time import sleep
from json import dumps
from kafka import KafkaProducer
# make sure you install kafka-python and not just kafka (pip3 install kafka-python)
import psutil
from datetime import datetime
kafka_server='localhost:9092'
time_interval = 1/2 # how often the data is checked in seconds (half second would be 1/2)
kafka_topic = 'numtest' #messages will be sent to that topic; you need to create this topic first
'''
Create a new topic numtest on Windows computer with this command:
C:\Program Files (x86)\MicroStrategy\Messaging Services\Kafka\kafka_2.13-3.4.0\bin\windows>kafka-topics.bat --bootstrap-server localhost:9092 --create --replication-factor 1 --partitions 1 --topic numtest
'''
computer_ID = 'XWAWX724' # any ID
producer = KafkaProducer(bootstrap_servers=[kafka_server],
value_serializer=lambda x:
dumps(x, indent=4, sort_keys=True, default=str).encode('utf-8'))
for e in range(10000):
data = {'message_id':e,
'computer_id':computer_ID,
'timestamp':datetime.now(),
'cpu_percent':psutil.cpu_percent(),
'virtual_memory_used':psutil.virtual_memory()[2],
'cpu_times_user':psutil.cpu_times()[0],
'swap_memory_used':psutil.swap_memory()[1]}
print(data)
producer.send(kafka_topic, value=data)
sleep(time_interval)
# run the program in CMD window with command:
# python C:\PythonScriptsLocation\KafkProd_real_name.py
### Kafka Consumer - it gets Kafka messages and forwards them into MSTR cubes
### It can push single messages or aggregate messages into bigger sets before pushing
### 2023-07-13 / Robert Prochowicz
# tested with Strategy 2021 update 9 and mstrio 11.3.10.101
# CHANGE8 - this tag marks lines that you need to edit when you want to duplicate the script for other Kafka topic
from kafka import KafkaConsumer
from json import loads
import pandas as pd
from mstrio.connection import Connection
from mstrio.project_objects import SuperCube
from datetime import datetime
import credentials_mstr #file with MSTR credentials
import pickle
### MSTRIO
base_url = 'http://YOUR_SERVER_MSTR/MicroStrategyLibrary/api'
login_mode = 1
project_id = 'B7CA92F04B9FAE8D941C3E9B7E0CD754'
mstr_folder = 'E3B770314B2D34031D0579A4E14BA576'
mstr_username=credentials_mstr.mstr_username
mstr_password=credentials_mstr.mstr_password
cube_name = "psutil2" #CHANGE8
cube_repository = "cube_repository.pkl"
to_metric_tup = ['cpu_percent','virtual_memory_used','cpu_times_user','swap_memory_used'] #CHANGE8
to_attrib_tup = ["message_id","computer_id","timestamp"] #CHANGE8
allowed_columns = to_metric_tup + to_attrib_tup
### KAFKA
kafka_topic = 'numtest' #CHANGE8
kafka_server = 'localhost:9092'
msg_package = 5 #how many messages should be gathered before pushing to MSTR? CHANGE8
### FUNCTIONS
def authenticate_mstr():
conn = Connection(base_url, mstr_username, mstr_password, project_id=project_id, login_mode=login_mode)
conn.connect()
return conn
def prepare_df(messg): # this will return a data frame // CHANGE8
df=pd.DataFrame.from_dict([messg.value])
# add kafka_timestamp from messg.timestamp
df['timestamp'] = df['timestamp'].astype('datetime64[ns]') #ocrrecting type of date column
for col in df.columns: #remove columns that you don't need
if col not in allowed_columns:
del df[col]
return df
def consumer_enable(kafka_server,kafka_topic):
consumer = KafkaConsumer(
kafka_topic,
bootstrap_servers=[kafka_server],
auto_offset_reset='latest',#earliest
enable_auto_commit=True,
group_id='my-group',
value_deserializer=lambda x: loads(x.decode('utf-8')))
return consumer
def create_cube(conn, df, cube_repository):
try:
dataset = SuperCube(connection=conn, name=cube_name)
dataset.add_table(name=cube_name, data_frame=df, update_policy="replace",
to_attribute=to_attrib_tup, to_metric=to_metric_tup)
dataset.create(folder_id=mstr_folder)
dataset_id = dataset.id # retrieve ID of the new cube
except:
print("Problem with connection to MSTR or with cube creation")
try:
f = open(cube_repository,"wb") # save cube ID to a file
pickle.dump(dataset_id,f)
f.close()
except:
print("Problem with creating a pickle file")
def update_cubes(conn,df,dataset_id):
try:
dataset = SuperCube(connection=conn, id=dataset_id)
dataset.add_table(name=cube_name, data_frame=df, update_policy="add")
dataset.update()
print("Data push successful")
except:
print("Problem with connection to MSTR")
def main():
choice = None
conn= authenticate_mstr()
consumer=consumer_enable(kafka_server,kafka_topic)
while choice != "0":
print \
("""
---MENU---B0292BB348C7FC1DB02635992475ECE9
0 - Exit
1 - Print message example
2 - First launch (cube creation)
3 - Enable Kafka Client and start pushing data to MSTR
""")
choice = input("Your choice: ")
if choice == "0":
print("Good bye!")
elif choice == "1":
print("Waiting for a message...")
for message in consumer:
print("Timestamp : " , message.timestamp)
print(message.value)
break
elif choice == "2":
conn.renew()
for message in consumer:
df = prepare_df(message)
create_cube(conn, df, cube_repository)
break
elif choice == "3":
f = open(cube_repository,"rb") # read saved cube IDs from file
dataset_id = pickle.load(f)
print(dataset_id)
f.close()
conn.renew()
msg_count=1
for message in consumer:
if msg_count==1: #initial creation of df
df = prepare_df(message)
else: #df created, now we catch messages
if msg_count % msg_package != 0: #number of collected messages < package size; collected, but not pushed to MSTR
df = pd.concat([prepare_df(message), df]) # add a new row to df
else: #package size reached - pushing to MSTR
df = pd.concat([prepare_df(message), df]) # add a new row to df
update_cubes(conn,df,dataset_id)
df = df.iloc[0:0] # purge df
msg_count += 1
else:
print(" ### Wrong option ### ")
main()