JSON 형식의 샘플 데이터를 Azure Data Explorer로 수집

이 문서에서는 Azure Data Explorer로 JSON 형식의 데이터를 수집하는 방법을 설명합니다. 원시 JSON 및 매핑된 JSON의 간단한 예에서 시작하여 여러 줄로된 JSON으로 계속 진행한 다음 배열과 사전을 포함하는 더 복잡한 JSON 스키마를 다룰 것입니다. 이 예제에서는 KQL(Kusto 쿼리 언어), C#또는 Python을 사용하여 JSON 형식의 데이터를 수집하는 프로세스를 자세히 설명합니다.

참고

프로덕션 시나리오에서는 관리 명령을 사용하지 .ingest 않는 것이 좋습니다. 대신 데이터 커넥터 를 사용하거나 Kusto 클라이언트 라이브러리 중 하나를 사용하여 프로그래밍 방식으로 데이터를 수집합니다.

사전 요구 사항

• Microsoft 계정 또는 Microsoft Entra 사용자 ID. Azure 구독이 필요하지 않습니다.
• Azure Data Explorer 클러스터 및 데이터베이스. 클러스터 및 데이터베이스를 만듭니다.

JSON 형식

Azure Data Explorer는 두 가지 JSON 파일 형식을 지원합니다.

• json: 줄로 구분된 JSON입니다. 입력 데이터의 각 줄에는 정확히 하나의 JSON 레코드가 있습니다. 이 형식은 주석 및 작은 따옴표로 묶인 속성의 구문 분석을 지원합니다. 자세한 내용은 JSON 줄을 참조하세요.
• multijson: 여러 줄로 된 JSON입니다. 파서는 줄 구분자를 무시하고 이전 위치에서 유효한 JSON의 끝까지 레코드를 읽습니다.

참고

수집 마법사를 사용하여 수집할 때 기본 형식은 입니다 multijson. 형식은 JSON 레코드의 여러 줄 JSON 레코드 및 배열을 처리할 수 있습니다. 구문 분석 오류가 발생하면 전체 파일이 삭제됩니다. 잘못된 JSON 레코드를 무시하려면 형식을 (JSON 줄)로 전환하는 json "데이터 형식 오류 무시" 옵션을 선택합니다.

JSON 선 형식()json을 사용하는 경우 구문 분석 중에 유효한 JSON 레코드를 나타내지 않는 줄을 건너뜁니다.

JSON 형식의 데이터 수집 및 매핑

JSON 형식의 데이터를 수집하려면 수집 속성을 사용하여 형식을 지정해야 합니다. JSON 데이터를 수집하려면 JSON 원본 항목을 대상 열에 매핑하는 매핑이 필요합니다. 데이터를 수집할 때 ingestionMappingReference(미리 정의된 매핑의 경우) 수집 속성 또는 IngestionMappings 속성과 함께 IngestionMapping 속성을 사용합니다. 이 문서에서는 수집에 사용된 테이블에 미리 정의된 ingestionMappingReference 수집 속성을 사용합니다. 아래 예에서는 JSON 레코드를 단일 열 테이블에 원시 데이터로 수집하는 것으로 시작합니다. 그런 다음 매핑을 사용하여 매핑된 열에 대한 각 속성을 수집합니다.

간단한 JSON 예

다음 예는 플랫 구조의 간단한 JSON입니다. 데이터에는 여러 디바이스에서 수집한 온도 및 습도 정보가 있습니다. 각 레코드는 ID와 타임스탬프로 표시됩니다.

{
    "timestamp": "2019-05-02 15:23:50.0369439",
    "deviceId": "2945c8aa-f13e-4c48-4473-b81440bb5ca2",
    "messageId": "7f316225-839a-4593-92b5-1812949279b3",
    "temperature": 31.0301639051317,
    "humidity": 62.0791099602725
}

원시 JSON 레코드 수집

이 예에서는 JSON 레코드를 단일 열 테이블에 원시 데이터로 수집합니다. 데이터 조작, 쿼리 사용 및 정책 업데이트는 데이터가 수집된 후에 수행됩니다.

KQL

Kusto 쿼리 언어 사용하여 원시 JSON 형식으로 데이터를 수집합니다.

1. [https://manage.visualstudio.com](https://dataexplorer.azure.com ) 에 로그인합니다.

2. 클러스터 추가를 선택합니다.

3. 클러스터 추가 대화 상자에 https://<ClusterName>.<Region>.kusto.windows.net/ 형식으로 클러스터 URL을 입력하고 추가를 선택합니다.

4. 다음 명령을 붙여넣고 실행을 선택하여 테이블을 만듭니다.

   .create table RawEvents (Event: dynamic)
   

   이 쿼리는 동적 데이터 유형의 단일 Event 열이 있는 테이블을 만듭니다.

5. JSON 매핑을 만듭니다.

   .create table RawEvents ingestion json mapping 'RawEventMapping' '[{"column":"Event","Properties":{"path":"$"}}]'
   

   이 명령은 매핑을 만들고 JSON 루트 경로 $를 Event 열에 매핑합니다.

6. 데이터를 RawEvents 테이블로 수집합니다.

   .ingest into table RawEvents ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json') with '{"format":"json", "ingestionMappingReference":"RawEventMapping"}'
   

C#

C#을 사용하여 원시 JSON 형식으로 데이터를 수집합니다.

1. RawEvents 테이블을 만듭니다.

   var kustoUri = "https://<clusterName>.<region>.kusto.windows.net/";
   var connectionStringBuilder = new KustoConnectionStringBuilder(kustoUri)
   {
       FederatedSecurity = true,
       UserID = userId,
       Password = password,
       Authority = tenantId,
       InitialCatalog = databaseName
   };
   using var kustoClient = KustoClientFactory.CreateCslAdminProvider(connectionStringBuilder);
   var tableName = "RawEvents";
   var command = CslCommandGenerator.GenerateTableCreateCommand(
       tableName,
       new[] { Tuple.Create("Events", "System.Object") }
   );
   await kustoClient.ExecuteControlCommandAsync(command);
   

2. JSON 매핑을 만듭니다.

   var tableMappingName = "RawEventMapping";
   command = CslCommandGenerator.GenerateTableMappingCreateCommand(
       IngestionMappingKind.Json,
       tableName,
       tableMappingName,
       new ColumnMapping[]
       {
           new() { ColumnName = "Events", Properties = new Dictionary<string, string> { { "path", "$" } } }
       }
   );
   
   await kustoClient.ExecuteControlCommandAsync(command);
   

   이 명령은 매핑을 만들고 JSON 루트 경로 $를 Event 열에 매핑합니다.

3. 데이터를 RawEvents 테이블로 수집합니다.

   var ingestUri = "https://ingest-<clusterName>.<region>.kusto.windows.net/";
   
   var ingestConnectionStringBuilder = new KustoConnectionStringBuilder(ingestUri)
   {
       FederatedSecurity = true,
       UserID = userId,
       Password = password,
       Authority = tenantId,
       InitialCatalog = databaseName
   };
   using var ingestClient = KustoIngestFactory.CreateQueuedIngestClient(ingestConnectionStringBuilder);
   
   var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json";
   var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
   {
       Format = DataSourceFormat.json,
       IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
   };
   await ingestClient.IngestFromStorageAsync(blobPath, properties);
   

참고

데이터는 일괄 처리 정책에 따라 집계되므로 몇 분의 대기 시간이 발생합니다.

Python

Python을 사용하여 원시 JSON 형식으로 데이터를 수집합니다.

1. RawEvents 테이블을 만듭니다.

   KUSTO_URI = "https://<ClusterName>.<Region>.kusto.windows.net/"
   KCSB_DATA = KustoConnectionStringBuilder.with_aad_device_authentication(KUSTO_URI, AAD_TENANT_ID)
   KUSTO_CLIENT = KustoClient(KCSB_DATA)
   TABLE = "RawEvents"
   
   CREATE_TABLE_COMMAND = ".create table " + TABLE + " (Events: dynamic)"
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_TABLE_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

2. JSON 매핑을 만듭니다.

   MAPPING = "RawEventMapping"
   CREATE_MAPPING_COMMAND = ".create table " + TABLE + " ingestion json mapping '" + MAPPING + """' '[{"column":"Event","path":"$"}]'"""
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_MAPPING_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

3. 데이터를 RawEvents 테이블로 수집합니다.

   INGEST_URI = "https://ingest-<ClusterName>.<Region>.kusto.windows.net/"
   KCSB_INGEST = KustoConnectionStringBuilder.with_aad_device_authentication(INGEST_URI, AAD_TENANT_ID)
   INGESTION_CLIENT = KustoIngestClient(KCSB_INGEST)
   BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json'
   
   INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.JSON, ingestion_mapping_reference=MAPPING)
   BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
   INGESTION_CLIENT.ingest_from_blob(
       BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)
   

   참고

   데이터는 일괄 처리 정책에 따라 집계되므로 몇 분의 대기 시간이 발생합니다.

매핑된 JSON 레코드 수집

이 예에서는 JSON 레코드 데이터를 수집합니다. 각 JSON 속성은 테이블의 단일 열에 매핑됩니다.

KQL

1. JSON 입력 데이터와 유사한 스키마를 사용하여 새 테이블을 만듭니다. 다음 모든 예와 수집 명령에 이 표를 사용합니다.

   .create table Events (Time: datetime, Device: string, MessageId: string, Temperature: double, Humidity: double)
   

2. JSON 매핑을 만듭니다.

   .create table Events ingestion json mapping 'FlatEventMapping' '[{"column":"Time","Properties":{"path":"$.timestamp"}},{"column":"Device","Properties":{"path":"$.deviceId"}},{"column":"MessageId","Properties":{"path":"$.messageId"}},{"column":"Temperature","Properties":{"path":"$.temperature"}},{"column":"Humidity","Properties":{"path":"$.humidity"}}]'
   

   이 매핑에서 테이블 스키마에 정의된 대로 timestamp 항목은 Time 열에 datetime 데이터 형식으로 수집됩니다.

3. 데이터를 Events 테이블로 수집합니다.

   .ingest into table Events ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json') with '{"format":"json", "ingestionMappingReference":"FlatEventMapping"}'
   

   'simple.json' 파일에는 몇 개의 줄로 구분된 JSON 레코드가 있습니다. 형식은 json이고 수집 명령에 사용된 매핑은 사용자가 만든 FlatEventMapping입니다.

C#

1. JSON 입력 데이터와 유사한 스키마를 사용하여 새 테이블을 만듭니다. 다음 모든 예와 수집 명령에 이 표를 사용합니다.

   var tableName = "Events";
   var command = CslCommandGenerator.GenerateTableCreateCommand(
      tableName,
      new[]
      {
          Tuple.Create("Time", "System.DateTime"),
          Tuple.Create("Device", "System.String"),
          Tuple.Create("MessageId", "System.String"),
          Tuple.Create("Temperature", "System.Double"),
          Tuple.Create("Humidity", "System.Double")
      }
   );
   await kustoClient.ExecuteControlCommandAsync(command);
   

2. JSON 매핑을 만듭니다.

   var tableMappingName = "FlatEventMapping";
   command = CslCommandGenerator.GenerateTableMappingCreateCommand(
       IngestionMappingKind.Json,
       tableName,
       tableMappingName,
       new ColumnMapping[]
       {
           new() { ColumnName = "Time", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.timestamp" } } },
           new() { ColumnName = "Device", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.deviceId" } } },
           new() { ColumnName = "MessageId", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.messageId" } } },
           new() { ColumnName = "Temperature", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.temperature" } } },
           new() { ColumnName = "Humidity", Properties = new Dictionary<string, string> { { MappingConsts.Path, "$.humidity" } } }
       }
   );
   await kustoClient.ExecuteControlCommandAsync(command);
   

   이 매핑에서 테이블 스키마에 정의된 대로 timestamp 항목은 Time 열에 datetime 데이터 형식으로 수집됩니다.

3. 데이터를 Events 테이블로 수집합니다.

   var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json";
   var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
   {
       Format = DataSourceFormat.json,
       IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
   };
   await ingestClient.IngestFromStorageAsync(blobPath, properties).ConfigureAwait(false);
   

   'simple.json' 파일에는 몇 개의 줄로 구분된 JSON 레코드가 있습니다. 형식은 json이고 수집 명령에 사용된 매핑은 사용자가 만든 FlatEventMapping입니다.

Python

1. JSON 입력 데이터와 유사한 스키마를 사용하여 새 테이블을 만듭니다. 다음 모든 예와 수집 명령에 이 표를 사용합니다.

   TABLE = "Events"
   CREATE_TABLE_COMMAND = ".create table " + TABLE + " (Time: datetime, Device: string, MessageId: string, Temperature: double, Humidity: double)"
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_TABLE_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

2. JSON 매핑을 만듭니다.

   MAPPING = "FlatEventMapping"
   CREATE_MAPPING_COMMAND = ".create table Events ingestion json mapping '" + MAPPING + """' '[{"column":"Time","Properties":{"path":"$.timestamp"}},{"column":"Device","Properties":{"path":"$.deviceId"}},{"column":"MessageId","Properties":{"path":"$.messageId"}},{"column":"Temperature","Properties":{"path":"$.temperature"}},{"column":"Humidity","Properties":{"path":"$.humidity"}}]'"""
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_MAPPING_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

3. 데이터를 Events 테이블로 수집합니다.

   BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json'
   
   INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.JSON, ingestion_mapping_reference=MAPPING)
   BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
   INGESTION_CLIENT.ingest_from_blob(
       BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)
   

   'simple.json' 파일에는 몇 개의 줄로 구분된 JSON 레코드가 있습니다. 형식은 json이고 수집 명령에 사용된 매핑은 사용자가 만든 FlatEventMapping입니다.

여러 줄로 된 JSON 레코드 수집

이 예에서는 여러 줄로 된 JSON 레코드를 수집합니다. 각 JSON 속성은 테이블의 단일 열에 매핑됩니다. 'multilines.json' 파일에는 들여쓰여진 JSON 레코드가 몇 개 있습니다. 형식 multijson 은 JSON 구조로 레코드를 읽는 것을 나타냅니다.

KQL

데이터를 Events 테이블로 수집합니다.

.ingest into table Events ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/multilined.json') with '{"format":"multijson", "ingestionMappingReference":"FlatEventMapping"}'

C#

데이터를 Events 테이블로 수집합니다.

var tableMappingName = "FlatEventMapping";
var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/multilined.json";
var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
{
    Format = DataSourceFormat.multijson,
    IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
};
await ingestClient.IngestFromStorageAsync(blobPath, properties).ConfigureAwait(false);

Python

데이터를 Events 테이블로 수집합니다.

MAPPING = "FlatEventMapping"
BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/multilined.json'
INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.MULTIJSON, ingestion_mapping_reference=MAPPING)
BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
INGESTION_CLIENT.ingest_from_blob(
    BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)

배열이 포함된 JSON 레코드 수집

배열 데이터 형식은 정렬된 값의 컬렉션입니다. JSON 배열의 수집은 업데이트 정책에 의해 수행됩니다. JSON은 있는 그대로 중간 테이블에 수집됩니다. 업데이트 정책은 RawEvents 테이블에서 미리 정의된 함수를 실행하여 결과를 대상 테이블로 다시 수집합니다. 다음 구조로 데이터를 수집합니다.

{
    "records":
    [
        {
            "timestamp": "2019-05-02 15:23:50.0000000",
            "deviceId": "ddbc1bf5-096f-42c0-a771-bc3dca77ac71",
            "messageId": "7f316225-839a-4593-92b5-1812949279b3",
            "temperature": 31.0301639051317,
            "humidity": 62.0791099602725
        },
        {
            "timestamp": "2019-05-02 15:23:51.0000000",
            "deviceId": "ddbc1bf5-096f-42c0-a771-bc3dca77ac71",
            "messageId": "57de2821-7581-40e4-861e-ea3bde102364",
            "temperature": 33.7529423105311,
            "humidity": 75.4787976739364
        }
    ]
}

KQL

1. mv-expand 연산자를 사용하여 컬렉션의 각 값이 별도의 행을 받도록 records 컬렉션을 확장하는 update policy 함수를 만듭니다. RawEvents 테이블을 원본 테이블로 사용하고 Events 테이블을 대상 테이블로 사용합니다.

   .create function EventRecordsExpand() {
       RawEvents
       | mv-expand records = Event.records
       | project
           Time = todatetime(records["timestamp"]),
           Device = tostring(records["deviceId"]),
           MessageId = tostring(records["messageId"]),
           Temperature = todouble(records["temperature"]),
           Humidity = todouble(records["humidity"])
   }
   

2. 함수가 수신한 스키마는 대상 테이블의 스키마와 일치해야 합니다. getschema 연산자를 사용하여 스키마를 검토합니다.

   EventRecordsExpand() | getschema
   

3. 대상 테이블에 업데이트 정책을 추가합니다. 이 정책은 RawEvents 중간 테이블에서 새로 수집된 데이터에 대한 쿼리를 자동으로 실행하고 결과를 Events 테이블로 수집합니다. 중간 테이블이 지속되는 것을 방지하기 위해 제로 보존 정책을 정의합니다.

   .alter table Events policy update @'[{"Source": "RawEvents", "Query": "EventRecordsExpand()", "IsEnabled": "True"}]'
   

4. 데이터를 RawEvents 테이블로 수집합니다.

   .ingest into table RawEvents ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/array.json') with '{"format":"multijson", "ingestionMappingReference":"RawEventMapping"}'
   

5. Events 테이블의 데이터를 검토합니다.

   Events
   

C#

1. mv-expand 연산자를 사용하여 컬렉션의 각 값이 별도의 행을 받도록 records 컬렉션을 확장하는 업데이트 함수를 만듭니다. RawEvents 테이블을 원본 테이블로 사용하고 Events 테이블을 대상 테이블로 사용합니다.

   var command = CslCommandGenerator.GenerateCreateFunctionCommand(
       "EventRecordsExpand",
       "UpdateFunctions",
       string.Empty,
       null,
       @"RawEvents
       | mv-expand records = Event
       | project
           Time = todatetime(records['timestamp']),
           Device = tostring(records['deviceId']),
           MessageId = tostring(records['messageId']),
           Temperature = todouble(records['temperature']),
           Humidity = todouble(records['humidity'])",
       ifNotExists: false
   );
   await kustoClient.ExecuteControlCommandAsync(command);
   

   참고

   함수가 수신한 스키마는 대상 테이블의 스키마와 일치해야 합니다.

2. 대상 테이블에 업데이트 정책을 추가합니다. 이 정책은 RawEvents 중간 테이블에서 새로 수집된 데이터에 대해 쿼리를 자동으로 실행하고 그 결과를 Events 테이블로 수집합니다. 중간 테이블이 지속되는 것을 방지하기 위해 제로 보존 정책을 정의합니다.

   command = ".alter table Events policy update @'[{'Source': 'RawEvents', 'Query': 'EventRecordsExpand()', 'IsEnabled': 'True'}]";
   await kustoClient.ExecuteControlCommandAsync(command);
   

3. 데이터를 RawEvents 테이블로 수집합니다.

   var blobPath = "https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/array.json";
   var tableName = "RawEvents";
   var tableMappingName = "RawEventMapping";
   var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
   {
       Format = DataSourceFormat.multijson,
       IngestionMapping = new IngestionMapping { IngestionMappingReference = tableMappingName }
   };
   await ingestClient.IngestFromStorageAsync(blobPath, properties);
   

4. Events 테이블의 데이터를 검토합니다.

Python

1. mv-expand 연산자를 사용하여 컬렉션의 각 값이 별도의 행을 받도록 records 컬렉션을 확장하는 업데이트 함수를 만듭니다. RawEvents 테이블을 원본 테이블로 사용하고 Events 테이블을 대상 테이블로 사용합니다.

   CREATE_FUNCTION_COMMAND =
       '''.create function EventRecordsExpand() {
           RawEvents
           | mv-expand records = Event
           | project
               Time = todatetime(records["timestamp"]),
               Device = tostring(records["deviceId"]),
               MessageId = tostring(records["messageId"]),
               Temperature = todouble(records["temperature"]),
               Humidity = todouble(records["humidity"])
           }'''
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_FUNCTION_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

   참고

   함수가 수신한 스키마는 대상 테이블의 스키마와 일치해야 합니다.

2. 대상 테이블에 업데이트 정책을 추가합니다. 이 정책은 RawEvents 중간 테이블에서 새로 수집된 데이터에 대해 쿼리를 자동으로 실행하고 그 결과를 Events 테이블로 수집합니다. 중간 테이블이 지속되는 것을 방지하기 위해 제로 보존 정책을 정의합니다.

   CREATE_UPDATE_POLICY_COMMAND =
       """.alter table Events policy update @'[{'Source': 'RawEvents', 'Query': 'EventRecordsExpand()', 'IsEnabled': 'True'}]"""
   RESPONSE = KUSTO_CLIENT.execute_mgmt(DATABASE, CREATE_UPDATE_POLICY_COMMAND)
   dataframe_from_result_table(RESPONSE.primary_results[0])
   

3. 데이터를 RawEvents 테이블로 수집합니다.

   TABLE = "RawEvents"
   MAPPING = "RawEventMapping"
   BLOB_PATH = 'https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/array.json'
   INGESTION_PROPERTIES = IngestionProperties(database=DATABASE, table=TABLE, dataFormat=DataFormat.MULTIJSON, ingestion_mapping_reference=MAPPING)
   BLOB_DESCRIPTOR = BlobDescriptor(BLOB_PATH, FILE_SIZE)
   INGESTION_CLIENT.ingest_from_blob(
       BLOB_DESCRIPTOR, ingestion_properties=INGESTION_PROPERTIES)
   

4. Events 테이블의 데이터를 검토합니다.

관련 콘텐츠

• 데이터 수집 개요
• 쿼리 작성