storing predictions within clickhouse

Author: brett-onions

src/routes/index.ts

@@ -277,9 +277,9 @@ async function getPrediction(trainingFileFormData: FormData, historicFutureFormD
     })
 
     logger.debug(`CHAP Prediction Results: ${prediction.status === 201 ? 'Successful Received Prediction':'Failed to Received Prediction'}`);
-
-    const stringifiedPrediction = JSON.stringify(prediction.data);
-    const originalFileName = 'prediction-result.json';
+    const { predictions } = prediction.data
+    const stringifiedPrediction = JSON.stringify(predictions);
+    const originalFileName = `prediction-result.json`;
     const fileUrl = await saveToTmp(Buffer.from(stringifiedPrediction), originalFileName);
 
     await uploadToMinio(
@@ -290,17 +290,6 @@ async function getPrediction(trainingFileFormData: FormData, historicFutureFormD
     );
     await fs.unlink(fileUrl);
 
-    //TODO: check if table does not exist
-    await client.exec({query: `CREATE TABLE ${bucket}_prediction (time_period String, predicted_value Float64) ENGINE = MergeTree ORDER BY (time_period);`});
-
-    await client.insert({
-      table: `${bucket}_prediction`,
-      values: prediction.data.predictions,
-      format: 'JSONEachRow'
-    });
-
-    logger.debug('Clickhouse Data Insertion Completed')
-
   } catch (error) {
     logger.error(`Failed to receive prediction: ${error}`);
   }

src/utils/clickhouse.ts

@@ -41,11 +41,22 @@ export async function createTable(fields: string[], tableName: string) {
   return true;
 }
 
+/**
+ * Create a table within clickhouse from the inferred schema from the json file
+ * if table already exists within the clickhouse function will return false
+ * 
+ * @param s3Path URL location of the json within Minio
+ * @param s3Config Access key and Secrete key credentials to access Minio
+ * @param tableName The name of the table to be created within Minio
+ * @param groupByColumnName The column the created table will be ORDERED By within clickhouse
+ * @returns 
+ */
+
 export async function createTableFromJson(
   s3Path: string,
   s3Config: { accessKey: string; secretKey: string },
   tableName: string,
-  key: string
+  groupByColumnName: string
 ) {
   const client = createClient({
     url,
@@ -54,19 +65,18 @@ export async function createTableFromJson(
 
   const normalizedTableName = tableName.replace(/-/g, '_');
 
-  //check if the table exists
   try {
     const existsResult = await client.query({
       query: `desc ${normalizedTableName}`,
     });
-    logger.info(`Table ${normalizedTableName} already exists`);
+    logger.debug(`Table ${normalizedTableName} already exists`);
     await client.close();
     return false;
   } catch (error) {
-    logger.error(`Table ${normalizedTableName} does not exist`);
+    logger.debug(`Table ${normalizedTableName} does not exist`);
   }
 
-  const query = generateDDLFromJson(s3Path, s3Config, normalizedTableName, key);
+  const query = generateDDLFromJson(s3Path, s3Config, normalizedTableName, groupByColumnName);
   await client.query({ query });
   await client.close();
 }
@@ -79,12 +89,12 @@ export function generateDDLFromJson(
   s3Path: string,
   s3Config: { accessKey: string; secretKey: string },
   tableName: string,
-  key: string
+  groupByColumnName: string
 ) {
   const query = `
   CREATE TABLE IF NOT EXISTS \`default\`.${tableName}
   ENGINE = MergeTree
-  ORDER BY ${key} EMPTY
+  ORDER BY ${groupByColumnName} EMPTY
   AS SELECT * 
   FROM s3('${s3Path}', '${s3Config.accessKey}', '${s3Config.secretKey}', JSONEachRow)
   SETTINGS schema_inference_make_columns_nullable = 0

src/utils/minioClient.ts

@@ -198,7 +198,7 @@ export async function createMinioBucketListeners(listOfBuckets: string[]) {
       const file = notification.s3.object.key;
 
       //@ts-ignore
-      const tableName = notification.s3.bucket.name;
+      const tableName = notification.s3.bucket.name + '_predictions';
 
       logger.info(`File received: ${file} from bucket ${tableName}`);
 
@@ -216,10 +216,10 @@ export async function createMinioBucketListeners(listOfBuckets: string[]) {
         if (extension === 'json' && validateJsonFile(fileBuffer)) {
           logger.debug('Now inserting ' + file + 'into clickhouse');
 
-          const key = getFirstField(JSON.parse(fileBuffer.toString()));
+          const groupByColumnName = getFirstField(JSON.parse(fileBuffer.toString()));
 
           // Create table from json
-          await createTableFromJson(minioUrl, { accessKey, secretKey }, tableName, key);
+          await createTableFromJson(minioUrl, { accessKey, secretKey }, tableName, groupByColumnName);
 
           // Insert data into clickhouse
           await insertFromS3Json(tableName, minioUrl, {