Fixed bug where nested json inside pandas wouldn't be ingested correctly

Author: AsafMah

azure-kusto-ingest/azure/kusto/ingest/base_ingest_client.py

@@ -1,22 +1,20 @@
+import gzip
 import ipaddress
 import os
 import tempfile
 import time
 import uuid
 from abc import ABCMeta, abstractmethod
-from copy import copy
 from enum import Enum
 from io import TextIOWrapper
 from typing import TYPE_CHECKING, Union, IO, AnyStr, Optional, Tuple
 from urllib.parse import urlparse
 
 from azure.kusto.data.data_format import DataFormat
 from azure.kusto.data.exceptions import KustoClosedError
-
 from .descriptors import FileDescriptor, StreamDescriptor
 from .ingestion_properties import IngestionProperties
 
-
 if TYPE_CHECKING:
     import pandas
 
@@ -117,12 +115,11 @@ def ingest_from_dataframe(self, df: "pandas.DataFrame", ingestion_properties: In
         if not isinstance(df, DataFrame):
             raise ValueError("Expected DataFrame instance, found {}".format(type(df)))
 
-        file_name = "df_{id}_{timestamp}_{uid}.csv.gz".format(id=id(df), timestamp=int(time.time()), uid=uuid.uuid4())
+        file_name = "df_{id}_{timestamp}_{uid}.json.gz".format(id=id(df), timestamp=int(time.time()), uid=uuid.uuid4())
         temp_file_path = os.path.join(tempfile.gettempdir(), file_name)
-
-        df.to_csv(temp_file_path, index=False, encoding="utf-8", header=False, compression="gzip")
-
-        ingestion_properties.format = DataFormat.CSV
+        with gzip.open(temp_file_path, "wb") as temp_file:
+            df.to_json(temp_file, orient="records", date_format="iso", lines=True)
+        ingestion_properties.format = DataFormat.JSON
 
         try:
             return self.ingest_from_file(temp_file_path, ingestion_properties)

azure-kusto-ingest/tests/test_e2e_ingest.py

@@ -529,15 +529,22 @@ async def test_streaming_ingest_from_dataframe(self):
             "xtextWithNulls",
             "xdynamicWithNulls",
         ]
+
+        guid = uuid.uuid4()
+
+        dynamic_value = ["me@dummy.com", "you@dummy.com", "them@dummy.com"]
         rows = [
-            [0, "00000000-0000-0000-0001-020304050607", 0.0, 0.0, 0, 0, 0, 0, 0, 0, 0, 0, "2014-01-01T01:01:01Z", "Zero", "Zero", "0", "00:00:00", None, ""]
+            [0, str(guid), 0.0, 0.0, 0, 0, 0, 0, 0, 0, 0, 0, "2014-01-01T01:01:01Z", "Zero", "Zero", "0", "00:00:00", None, dynamic_value]
         ]
         df = DataFrame(data=rows, columns=fields)
         ingestion_properties = IngestionProperties(database=self.test_db, table=self.test_table, flush_immediately=True, data_format=DataFormat.CSV)
         self.ingest_client.ingest_from_dataframe(df, ingestion_properties)
 
         await self.assert_rows_added(1, timeout=120)
 
+        a = self.client.execute(self.test_db, f"{self.test_table} | where rowguid == '{guid}'")
+        assert a.primary_results[0].rows[0]["xdynamicWithNulls"] == dynamic_value
+
     @pytest.mark.asyncio
     async def test_streaming_ingest_from_blob(self, is_managed_streaming):
         ingestion_properties = IngestionProperties(