Integration with MQTT

yongwei

Currently, I was trying to integrate Vertica with MQTT, I tried to follow the technical exploration tutorial below but I get stuck on step 3 executing the copy_all_files_to_vertica.py. I notice that the test environment is using Vertica 11 while in my development environment is Vertica9.

https://www.vertica.com/kb/MQTT_TE/Content/Partner/MQTT_TE.htm

The error I received was data_0_completed.JSON
Connection closed by Vertica
Traceback (most recent call last):
File "mqtt/copy_all_files_to_vertica.py", line 43, in
get_all_the_files_list_copy_to_vertica()
File "mqtt/copy_all_files_to_vertica.py", line 33, in get_all_the_files_list_copy_to_vertica
dbObj = DatabaseManager()
File "mqtt/copy_all_files_to_vertica.py", line 11, in init
self.conn.commit()
AttributeError: 'DatabaseManager' object has no attribute 'conn'
Exception ignored in: >
Traceback (most recent call last):
File "mqtt/copy_all_files_to_vertica.py", line 24, in del
self.cur.close()
AttributeError: 'DatabaseManager' object has no attribute 'cur'

Is it a product limitation on Vertica 9 or is it possible for me to integrate MQTT with Vertica 9?

Hemanthrahul

Hello Yongwei,

Can you also provide us the code you have used in the DatabaseManager class in the copy_all_files_to_vertica.py?

and you able to connect to your vertica 9 instance successfully from other tools?

Thanks.

RajasekharVuppala

Hi Yongwei,

We have integrated MQTT with Vertica 11 and Today I have Tested MQTT with Vertica 12.0.1 and it's working fine.

Can you try to upgrade to Vertica 11 or latest version(recommended) of Vertica?

I Hope your Vertica9 is working fine(You can check via VSQL) and you have entered Correct Vertica connection details in database manager class?

Thanks,
Rajasekhar Vuppala

yongwei

Hi, sure I will test it out tomorrow. this is the code that I'm using.
import vertica_python
import os

class DatabaseManager():
def init(self):
conn_info = {'host': 'host.com', 'port': 5450, 'user': 'dbadmin', 'password': 'password', 'database': 'PREDICTIVEMAINT'}
try:
self.conn = vertica_python.connect(**conn_info)
except Exception as e:
print(e)
self.conn.commit()
self.cur = self.conn.cursor()

def add_del_update_db_record(self, sql_query, args=()):
    self.cur.execute(sql_query, args)
    self.conn.commit()
    return
def copy_JSON_records_to_vertica(self, sql_query, args=()):
    self.cur.copy(sql_query, args)
    self.conn.commit()
    return

def __del__(self):
    self.cur.close()
    self.conn.close()

The below function will be continuosly looking for JSON files with "completed" name in them

def get_all_the_files_list_copy_to_vertica():
while (True):
for i in os.listdir():
if "completed" in i:
print(i)
dbObj = DatabaseManager()
with open(i, "rb") as fs:
dbObj.copy_JSON_records_to_vertica("COPY mqtt.Sensor_Data( DeviceID, pH, EC, Latitude, Longtitude, DateTime) FROM STDIN parser fjsonparser()", fs)
del dbObj
print ("Inserted Sensor Data into Database.")
print ("")
fs.close()
os.remove(i)

get_all_the_files_list_copy_to_vertica()

RajasekharVuppala

Hi @yongwei ,

Code looks good.

Can you update the Vertica to Vertica 11 or latest version of Vertica.

Please let us know how it goes.

yongwei

Hi,
thanks for the support, the code is working now in vertica 9 and previously it doesn't work because i put the wrong credentials