Spark schemaEvolution의 실제 사용

독자 여러분, 좋은 하루 되세요!

이 기사에서는 Neoflex 빅 데이터 솔루션 비즈니스 영역의 수석 컨설턴트가 Apache Spark를 사용하여 가변 구조 쇼케이스를 구축하는 옵션에 대해 자세히 설명합니다.

데이터 분석 프로젝트의 일환으로 느슨하게 구조화된 데이터를 기반으로 매장을 구축하는 작업이 자주 발생합니다.

일반적으로 JSON 또는 XML로 저장된 다양한 시스템의 로그 또는 응답입니다. 데이터가 Hadoop에 업로드된 다음 데이터에서 매장을 구축해야 합니다. 예를 들어 Impala를 통해 생성된 쇼케이스에 대한 액세스를 구성할 수 있습니다.

이 경우 대상 상점 첫화면의 스키마를 미리 알 수 없습니다. 더욱이 체계는 데이터에 의존하기 때문에 미리 작성할 수 없으며 우리는 이러한 매우 느슨하게 구조화된 데이터를 다루고 있습니다.

예를 들어 오늘 다음 응답이 기록됩니다.

{source: "app1", error_code: ""}

내일 같은 시스템에서 다음과 같은 답변이 나옵니다.

{source: "app1", error_code: "error", description: "Network error"}

결과적으로 쇼케이스에 설명 필드가 하나 더 추가되어야 하며 그것이 올지 여부는 아무도 모릅니다.

그러한 데이터에 상점 첫화면을 생성하는 작업은 매우 표준적이며 Spark에는 이를 위한 여러 도구가 있습니다. 소스 데이터를 구문 분석하기 위해 JSON과 XML을 모두 지원하며 이전에 알려지지 않은 스키마에 대해서는 schemaEvolution을 지원합니다.

언뜻 보기에 해결책은 간단해 보입니다. JSON이 포함된 폴더를 가져와서 데이터 프레임으로 읽어야 합니다. Spark는 스키마를 만들고 중첩된 데이터를 구조로 바꿉니다. 또한 Hive 메타스토어에 매장을 등록하여 Impala에서도 지원되는 마루에 모든 것을 저장해야 합니다.

모든 것이 단순한 것 같습니다.

그러나 문서의 짧은 예제에서는 실제로 많은 문제를 어떻게 처리해야 하는지 명확하지 않습니다.

문서는 상점 첫 화면을 작성하는 것이 아니라 JSON 또는 XML을 데이터 프레임으로 읽는 접근 방식을 설명합니다.

즉, 단순히 JSON을 읽고 구문 분석하는 방법을 보여줍니다.

df = spark.read.json(path...)

이는 Spark에서 데이터를 사용할 수 있도록 만들기에 충분합니다.

실제로 스크립트는 폴더에서 JSON 파일을 읽고 데이터 프레임을 만드는 것보다 훨씬 더 복잡합니다. 상황은 다음과 같습니다. 이미 특정 매장이 있고 매일 새로운 데이터가 들어오고 매장에 추가해야하며 계획이 다를 수 있음을 잊지 마십시오.

쇼케이스를 구축하는 일반적인 계획은 다음과 같습니다.

1 단계. 데이터는 후속 일일 다시 로드와 함께 Hadoop에 로드되고 새 파티션에 추가됩니다. 날짜별로 분할된 초기 데이터가 있는 폴더가 나타납니다.

2 단계. 초기 로드 중에 Spark에서 이 폴더를 읽고 구문 분석합니다. 결과 데이터 프레임은 구문 분석 가능한 형식(예: Parquet 형식)으로 저장되어 Impala로 가져올 수 있습니다. 이렇게 하면 지금까지 누적된 모든 데이터로 대상 쇼케이스가 생성됩니다.

3 단계. 매일 스토어프론트를 업데이트하는 다운로드가 생성됩니다.
증분 로드, 쇼케이스를 분할해야 할 필요성, 쇼케이스의 일반 구성 유지 문제가 있습니다.

예를 들어 보겠습니다. 리포지토리 구축의 첫 번째 단계가 구현되었고 JSON 파일이 폴더에 업로드되었다고 가정해 보겠습니다.

그들로부터 데이터 프레임을 생성한 다음 쇼케이스로 저장하는 것은 문제가 되지 않습니다. 이것은 Spark 설명서에서 쉽게 찾을 수 있는 첫 번째 단계입니다.

df = spark.read.option("mergeSchema", True).json(".../*") 
df.printSchema()

root 
|-- a: long (nullable = true) 
|-- b: string (nullable = true) 
|-- c: struct (nullable = true) |    
|-- d: long (nullable = true)

모든 것이 괜찮은 것 같습니다.

JSON을 읽고 구문 분석한 다음 데이터 프레임을 쪽모이 세공으로 저장하고 편리한 방법으로 Hive에 등록합니다.

df.write.format(“parquet”).option('path','<External Table Path>').saveAsTable('<Table Name>')

우리는 창문을 얻습니다.

그러나 다음날 소스의 새 데이터가 추가되었습니다. JSON이 포함된 폴더와 이 폴더에서 생성된 쇼케이스가 있습니다. 원본에서 다음 데이터 배치를 로드한 후 데이터 마트에서 하루 분량의 데이터가 누락되었습니다.

논리적 솔루션은 점포를 날짜별로 분할하여 다음 날마다 새 분할을 추가할 수 있도록 하는 것입니다. 이에 대한 메커니즘도 잘 알려져 있으며 Spark를 사용하면 파티션을 별도로 작성할 수 있습니다.

먼저 초기 로드를 수행하여 위에서 설명한 대로 데이터를 저장하고 분할만 추가합니다. 이 작업을 매장 초기화라고 하며 한 번만 수행됩니다.

df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable)

다음날 새 파티션만 로드합니다.

df.coalesce(1).write.mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable +"/date_load=" + date_load + "/")

남은 것은 Hive에 다시 등록하여 스키마를 업데이트하는 것입니다.
그러나 여기서 문제가 발생합니다.

첫 번째 문제. 조만간 결과 쪽모이 세공 마루를 읽을 수 없게 됩니다. 이는 Parquet와 JSON이 빈 필드를 다르게 처리하는 방식 때문입니다.

일반적인 상황을 생각해 봅시다. 예를 들어 어제 JSON이 도착했습니다.

День 1: {"a": {"b": 1}},

오늘날 동일한 JSON은 다음과 같습니다.

День 2: {"a": null}

각각 한 줄이 있는 두 개의 다른 파티션이 있다고 가정해 보겠습니다.
전체 소스 데이터를 읽을 때 Spark는 유형을 결정할 수 있으며 "a"가 "구조" 유형의 필드이고 INT 유형의 중첩 필드 "b"가 있음을 이해할 것입니다. 그러나 각 파티션이 개별적으로 저장된 경우 호환되지 않는 파티션 구성표가 있는 마루가 생성됩니다.

df1 (a: <struct<"b": INT>>)
df2 (a: STRING NULLABLE)

이 상황은 잘 알려져 있으므로 소스 데이터를 구문 분석할 때 빈 필드를 제거하는 옵션이 특별히 추가되었습니다.

df = spark.read.json("...", dropFieldIfAllNull=True)

이 경우 마루는 함께 읽을 수 있는 파티션으로 구성됩니다.
실제로 이것을 해본 사람들은 여기서 씁쓸하게 웃을 것입니다. 왜? 예, 두 가지 상황이 더 있을 가능성이 높기 때문입니다. 또는 세. 또는 1. 거의 확실하게 발생하는 첫 번째는 숫자 유형이 다른 JSON 파일에서 다르게 보인다는 것입니다. 예: {intField: 1.1} 및 {intField: XNUMX}. 이러한 필드가 하나의 파티션에서 발견되면 스키마 병합이 모든 것을 올바르게 읽어 가장 정확한 유형으로 이어집니다. 그러나 다른 경우 하나는 intField: int이고 다른 하나는 intField: double입니다.

이 상황을 처리하기 위한 다음 플래그가 있습니다.

df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)

이제 단일 데이터 프레임으로 읽을 수 있는 파티션과 전체 쇼케이스의 유효한 마루가 있는 폴더가 있습니다. 예? 아니요.

Hive에 테이블을 등록했음을 기억해야 합니다. Hive는 필드 이름에서 대소문자를 구분하지 않지만 Parquet은 대소문자를 구분합니다. 따라서 스키마가 있는 파티션: field1: int 및 Field1: int는 Hive에서 동일하지만 Spark에서는 동일하지 않습니다. 필드 이름을 소문자로 변환하는 것을 잊지 마십시오.

그 후에는 모든 것이 괜찮은 것 같습니다.

그러나 모든 것이 그렇게 간단하지는 않습니다. 잘 알려진 두 번째 문제가 있습니다. 각각의 새 파티션은 별도로 저장되므로 파티션 폴더에는 Spark 서비스 파일(예: _SUCCESS 작업 성공 플래그)이 포함됩니다. 이로 인해 쪽모이 세공을 시도할 때 오류가 발생합니다. 이를 방지하려면 Spark가 폴더에 서비스 파일을 추가하지 못하도록 구성을 구성해야 합니다.

hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")

이제 매일 새로운 쪽모이 세공 파티션이 대상 쇼케이스 폴더에 추가되는 것 같습니다. 이 폴더에는 그날의 파싱된 데이터가 있습니다. 데이터 유형이 충돌하는 파티션이 없는지 미리 확인했습니다.

그러나 세 번째 문제가 있습니다. 이제 일반 스키마는 알려지지 않았으며, Hive의 테이블에는 잘못된 스키마가 있습니다. 각각의 새 파티션이 스키마에 왜곡을 도입했을 가능성이 높기 때문입니다.

테이블을 다시 등록해야 합니다. 이는 간단하게 수행할 수 있습니다. 상점 첫화면의 마루를 다시 읽고, 스키마를 가져오고 이를 기반으로 DDL을 생성하여 Hive의 폴더를 외부 테이블로 다시 등록하고 대상 상점 첫화면의 스키마를 업데이트합니다.

네 번째 문제가 있습니다. 테이블을 처음 등록할 때 Spark에 의존했습니다. 이제 우리는 직접 작업을 수행하며 쪽모이 세공 마루 필드는 Hive에 허용되지 않는 문자로 시작할 수 있음을 기억해야 합니다. 예를 들어 Spark는 "corrupt_record" 필드에서 구문 분석할 수 없는 줄을 버립니다. 이러한 필드는 이스케이프하지 않고는 Hive에 등록할 수 없습니다.

이것을 알면 다음과 같은 계획을 얻습니다.

f_def = ""
for f in pf.dtypes:
  if f[0] != "date_load":
    f_def = f_def + "," + f[0].replace("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") 
table_define = "CREATE EXTERNAL TABLE jsonevolvtable (" + f_def[1:] + " ) "
table_define = table_define + "PARTITIONED BY (date_load string) STORED AS PARQUET LOCATION '/user/admin/testJson/testSchemaEvolution/pq/'"
hc.sql("drop table if exists jsonevolvtable")
hc.sql(table_define)

암호 ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("배열<`", "배열<") 다음 대신 안전한 DDL을 만듭니다.

create table tname (_field1 string, 1field string)

"_field1, 1field"와 같은 필드 이름을 사용하면 필드 이름이 이스케이프되는 안전한 DDL이 만들어집니다: create table `tname` (`_field1` 문자열, `1field` 문자열).

질문이 생깁니다. 완전한 스키마(pf 코드)로 데이터 프레임을 올바르게 가져오는 방법은 무엇입니까? 이 pf를 얻는 방법? 다섯 번째 문제입니다. 대상 쇼케이스의 쪽모이 세공 파일이 있는 폴더에서 모든 파티션 구성표를 다시 읽으시겠습니까? 이 방법은 가장 안전하지만 어렵습니다.

스키마는 이미 Hive에 있습니다. 전체 테이블의 스키마와 새 파티션을 결합하여 새 스키마를 얻을 수 있습니다. 따라서 Hive에서 테이블 스키마를 가져와 새 파티션의 스키마와 결합해야 합니다. Hive에서 테스트 메타데이터를 읽고 임시 폴더에 저장한 다음 Spark를 사용하여 한 번에 두 파티션을 읽으면 됩니다.

실제로 Hive의 원래 테이블 스키마와 새 파티션 등 필요한 모든 것이 있습니다. 데이터도 있습니다. 상점 첫화면 스키마와 생성된 파티션의 새 필드를 결합하는 새 스키마를 가져오기만 하면 됩니다.

from pyspark.sql import HiveContext
from pyspark.sql.functions import lit
hc = HiveContext(spark)
df = spark.read.json("...", dropFieldIfAllNull=True)
df.write.mode("overwrite").parquet(".../date_load=12-12-2019")
pe = hc.sql("select * from jsonevolvtable limit 1")
pe.write.mode("overwrite").parquet(".../fakePartiton/")
pf = spark.read.option("mergeSchema", True).parquet(".../date_load=12-12-2019/*", ".../fakePartiton/*")

다음으로 이전 스니펫에서와 같이 테이블 등록 DDL을 생성합니다.
전체 체인이 올바르게 작동하면, 즉 초기화 로드가 있고 테이블이 Hive에서 올바르게 생성되면 업데이트된 테이블 스키마를 얻습니다.

그리고 마지막 문제는 Hive 테이블에 파티션을 추가할 수 없다는 것입니다. 파티션이 깨질 것이기 때문입니다. Hive가 파티션 구조를 수정하도록 강제해야 합니다.

from pyspark.sql import HiveContext
hc = HiveContext(spark) 
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)

JSON을 읽고 이를 기반으로 상점 첫 화면을 만드는 간단한 작업으로 여러 암묵적인 어려움을 극복할 수 있으며, 이에 대한 솔루션은 별도로 찾아야 합니다. 이러한 솔루션은 간단하지만 찾는 데 많은 시간이 걸립니다.

쇼케이스 구성을 구현하려면 다음을 수행해야 했습니다.

• 쇼케이스에 파티션 추가, 서비스 파일 제거
• Spark가 입력한 소스 데이터의 빈 필드 처리
• 단순 유형을 문자열로 캐스트
• 필드 이름을 소문자로 변환
• Hive에서 별도의 데이터 업로드 및 테이블 등록(DDL 생성)
• Hive와 호환되지 않을 수 있는 필드 이름을 이스케이프 처리하는 것을 잊지 마십시오.
• Hive에서 테이블 등록을 업데이트하는 방법 알아보기

요약하면 상점 창을 짓기로 한 결정에는 많은 함정이 있습니다. 따라서 구현이 어려운 경우 성공적인 전문 지식을 갖춘 경험이 풍부한 파트너에게 문의하는 것이 좋습니다.

이 기사를 읽어 주셔서 감사합니다. 유용한 정보가 되셨기를 바랍니다.

출처 : habr.com