ਅਭਿਆਸ ਵਿੱਚ ਸਪਾਰਕ ਸਕੀਮਾ ਈਵੇਲੂਸ਼ਨ

ਪਿਆਰੇ ਪਾਠਕੋ, ਚੰਗਾ ਦਿਨ!

ਇਸ ਲੇਖ ਵਿੱਚ, Neoflex ਦੇ Big Data Solutions ਕਾਰੋਬਾਰੀ ਖੇਤਰ ਦੇ ਪ੍ਰਮੁੱਖ ਸਲਾਹਕਾਰ ਅਪਾਚੇ ਸਪਾਰਕ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ ਵੇਰੀਏਬਲ ਸਟ੍ਰਕਚਰ ਸ਼ੋਅਕੇਸ ਬਣਾਉਣ ਦੇ ਵਿਕਲਪਾਂ ਦਾ ਵਿਸਥਾਰ ਵਿੱਚ ਵਰਣਨ ਕਰਦੇ ਹਨ।

ਇੱਕ ਡੇਟਾ ਵਿਸ਼ਲੇਸ਼ਣ ਪ੍ਰੋਜੈਕਟ ਦੇ ਹਿੱਸੇ ਵਜੋਂ, ਢਿੱਲੇ ਢਾਂਚੇ ਵਾਲੇ ਡੇਟਾ ਦੇ ਅਧਾਰ ਤੇ ਸਟੋਰਫਰੰਟ ਬਣਾਉਣ ਦਾ ਕੰਮ ਅਕਸਰ ਉੱਠਦਾ ਹੈ।

ਆਮ ਤੌਰ 'ਤੇ ਇਹ ਲੌਗ, ਜਾਂ ਵੱਖ-ਵੱਖ ਸਿਸਟਮਾਂ ਤੋਂ ਜਵਾਬ ਹੁੰਦੇ ਹਨ, ਜੋ JSON ਜਾਂ XML ਵਜੋਂ ਸੁਰੱਖਿਅਤ ਕੀਤੇ ਜਾਂਦੇ ਹਨ। ਡਾਟਾ Hadoop 'ਤੇ ਅੱਪਲੋਡ ਕੀਤਾ ਗਿਆ ਹੈ, ਫਿਰ ਤੁਹਾਨੂੰ ਉਹਨਾਂ ਤੋਂ ਇੱਕ ਸਟੋਰਫਰੰਟ ਬਣਾਉਣ ਦੀ ਲੋੜ ਹੈ। ਅਸੀਂ ਬਣਾਏ ਗਏ ਸ਼ੋਕੇਸ ਤੱਕ ਪਹੁੰਚ ਦਾ ਪ੍ਰਬੰਧ ਕਰ ਸਕਦੇ ਹਾਂ, ਉਦਾਹਰਨ ਲਈ, Impala ਦੁਆਰਾ।

ਇਸ ਸਥਿਤੀ ਵਿੱਚ, ਟਾਰਗੇਟ ਸਟੋਰਫਰੰਟ ਦੀ ਸਕੀਮਾ ਪਹਿਲਾਂ ਤੋਂ ਜਾਣੀ ਨਹੀਂ ਜਾਂਦੀ. ਇਸ ਤੋਂ ਇਲਾਵਾ, ਸਕੀਮ ਨੂੰ ਵੀ ਪਹਿਲਾਂ ਤੋਂ ਤਿਆਰ ਨਹੀਂ ਕੀਤਾ ਜਾ ਸਕਦਾ, ਕਿਉਂਕਿ ਇਹ ਡੇਟਾ 'ਤੇ ਨਿਰਭਰ ਕਰਦਾ ਹੈ, ਅਤੇ ਅਸੀਂ ਇਹਨਾਂ ਬਹੁਤ ਹੀ ਢਿੱਲੇ ਢਾਂਚੇ ਵਾਲੇ ਡੇਟਾ ਨਾਲ ਨਜਿੱਠ ਰਹੇ ਹਾਂ।

ਉਦਾਹਰਨ ਲਈ, ਅੱਜ ਹੇਠਾਂ ਦਿੱਤਾ ਜਵਾਬ ਲੌਗ ਕੀਤਾ ਗਿਆ ਹੈ:

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

ਅਤੇ ਕੱਲ੍ਹ ਉਸੇ ਸਿਸਟਮ ਤੋਂ ਹੇਠਾਂ ਦਿੱਤਾ ਜਵਾਬ ਆਵੇਗਾ:

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

ਨਤੀਜੇ ਵਜੋਂ, ਇੱਕ ਹੋਰ ਖੇਤਰ ਨੂੰ ਸ਼ੋਅਕੇਸ ਵਿੱਚ ਜੋੜਿਆ ਜਾਣਾ ਚਾਹੀਦਾ ਹੈ - ਵਰਣਨ, ਅਤੇ ਕੋਈ ਨਹੀਂ ਜਾਣਦਾ ਕਿ ਇਹ ਆਵੇਗਾ ਜਾਂ ਨਹੀਂ.

ਅਜਿਹੇ ਡੇਟਾ 'ਤੇ ਸਟੋਰਫਰੰਟ ਬਣਾਉਣ ਦਾ ਕੰਮ ਬਹੁਤ ਮਿਆਰੀ ਹੈ, ਅਤੇ ਸਪਾਰਕ ਕੋਲ ਇਸਦੇ ਲਈ ਬਹੁਤ ਸਾਰੇ ਸਾਧਨ ਹਨ. ਸਰੋਤ ਡੇਟਾ ਨੂੰ ਪਾਰਸ ਕਰਨ ਲਈ, JSON ਅਤੇ XML ਦੋਵਾਂ ਲਈ ਸਮਰਥਨ ਹੈ, ਅਤੇ ਪਹਿਲਾਂ ਅਣਜਾਣ ਸਕੀਮਾ ਲਈ, ਸਕੀਮਾ ਈਵੋਲੂਸ਼ਨ ਲਈ ਸਮਰਥਨ ਪ੍ਰਦਾਨ ਕੀਤਾ ਗਿਆ ਹੈ।

ਪਹਿਲੀ ਨਜ਼ਰ 'ਤੇ, ਹੱਲ ਸਧਾਰਨ ਲੱਗਦਾ ਹੈ. ਤੁਹਾਨੂੰ JSON ਦੇ ਨਾਲ ਇੱਕ ਫੋਲਡਰ ਲੈਣ ਅਤੇ ਇਸਨੂੰ ਡੇਟਾਫ੍ਰੇਮ ਵਿੱਚ ਪੜ੍ਹਨ ਦੀ ਲੋੜ ਹੈ। ਸਪਾਰਕ ਇੱਕ ਸਕੀਮਾ ਬਣਾਏਗਾ, ਨੇਸਟਡ ਡੇਟਾ ਨੂੰ ਢਾਂਚੇ ਵਿੱਚ ਬਦਲ ਦੇਵੇਗਾ। ਇਸ ਤੋਂ ਇਲਾਵਾ, ਹਰ ਚੀਜ਼ ਨੂੰ ਪਾਰਕਵੇਟ ਵਿੱਚ ਸੁਰੱਖਿਅਤ ਕਰਨ ਦੀ ਜ਼ਰੂਰਤ ਹੈ, ਜੋ ਕਿ ਇਮਪਾਲਾ ਵਿੱਚ ਵੀ ਸਮਰਥਿਤ ਹੈ, Hive ਮੇਟਾਸਟੋਰ ਵਿੱਚ ਸਟੋਰਫਰੰਟ ਨੂੰ ਰਜਿਸਟਰ ਕਰਕੇ।

ਹਰ ਚੀਜ਼ ਸਧਾਰਨ ਜਾਪਦੀ ਹੈ.

ਹਾਲਾਂਕਿ, ਦਸਤਾਵੇਜ਼ ਵਿੱਚ ਛੋਟੀਆਂ ਉਦਾਹਰਣਾਂ ਤੋਂ ਇਹ ਸਪੱਸ਼ਟ ਨਹੀਂ ਹੈ ਕਿ ਅਭਿਆਸ ਵਿੱਚ ਕਈ ਸਮੱਸਿਆਵਾਂ ਦਾ ਕੀ ਕਰਨਾ ਹੈ।

ਦਸਤਾਵੇਜ਼ ਇੱਕ ਸਟੋਰਫਰੰਟ ਬਣਾਉਣ ਲਈ ਨਹੀਂ, ਪਰ ਇੱਕ ਡੇਟਾਫ੍ਰੇਮ ਵਿੱਚ JSON ਜਾਂ XML ਨੂੰ ਪੜ੍ਹਨ ਲਈ ਇੱਕ ਪਹੁੰਚ ਦਾ ਵਰਣਨ ਕਰਦਾ ਹੈ।

ਅਰਥਾਤ, ਇਹ ਸਿਰਫ਼ ਇਹ ਦਿਖਾਉਂਦਾ ਹੈ ਕਿ JSON ਨੂੰ ਕਿਵੇਂ ਪੜ੍ਹਨਾ ਅਤੇ ਪਾਰਸ ਕਰਨਾ ਹੈ:

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

ਇਹ ਸਪਾਰਕ ਨੂੰ ਡੇਟਾ ਉਪਲਬਧ ਕਰਾਉਣ ਲਈ ਕਾਫੀ ਹੈ।

ਅਭਿਆਸ ਵਿੱਚ, ਸਕ੍ਰਿਪਟ ਇੱਕ ਫੋਲਡਰ ਤੋਂ JSON ਫਾਈਲਾਂ ਨੂੰ ਪੜ੍ਹਨ ਅਤੇ ਇੱਕ ਡੇਟਾਫ੍ਰੇਮ ਬਣਾਉਣ ਨਾਲੋਂ ਬਹੁਤ ਜ਼ਿਆਦਾ ਗੁੰਝਲਦਾਰ ਹੈ. ਸਥਿਤੀ ਇਸ ਤਰ੍ਹਾਂ ਦਿਖਾਈ ਦਿੰਦੀ ਹੈ: ਪਹਿਲਾਂ ਹੀ ਇੱਕ ਖਾਸ ਸਟੋਰਫਰੰਟ ਹੈ, ਹਰ ਰੋਜ਼ ਨਵਾਂ ਡੇਟਾ ਆਉਂਦਾ ਹੈ, ਉਹਨਾਂ ਨੂੰ ਸਟੋਰਫਰੰਟ ਵਿੱਚ ਜੋੜਨ ਦੀ ਜ਼ਰੂਰਤ ਹੁੰਦੀ ਹੈ, ਇਹ ਨਾ ਭੁੱਲੋ ਕਿ ਸਕੀਮ ਵੱਖਰੀ ਹੋ ਸਕਦੀ ਹੈ.

ਸ਼ੋਅਕੇਸ ਬਣਾਉਣ ਲਈ ਆਮ ਸਕੀਮ ਹੇਠ ਲਿਖੇ ਅਨੁਸਾਰ ਹੈ:

1 ਕਦਮ. ਡੇਟਾ ਨੂੰ ਬਾਅਦ ਵਿੱਚ ਰੋਜ਼ਾਨਾ ਰੀਲੋਡਿੰਗ ਨਾਲ Hadoop ਵਿੱਚ ਲੋਡ ਕੀਤਾ ਜਾਂਦਾ ਹੈ ਅਤੇ ਇੱਕ ਨਵੇਂ ਭਾਗ ਵਿੱਚ ਜੋੜਿਆ ਜਾਂਦਾ ਹੈ। ਇਹ ਦਿਨ ਦੁਆਰਾ ਵਿਭਾਜਿਤ ਸ਼ੁਰੂਆਤੀ ਡੇਟਾ ਦੇ ਨਾਲ ਇੱਕ ਫੋਲਡਰ ਬਣ ਜਾਂਦਾ ਹੈ।

2 ਕਦਮ. ਸ਼ੁਰੂਆਤੀ ਲੋਡ ਦੌਰਾਨ, ਇਸ ਫੋਲਡਰ ਨੂੰ ਸਪਾਰਕ ਦੁਆਰਾ ਪੜ੍ਹਿਆ ਅਤੇ ਪਾਰਸ ਕੀਤਾ ਜਾਂਦਾ ਹੈ। ਨਤੀਜਾ ਡੇਟਾਫ੍ਰੇਮ ਪਾਰਸਯੋਗ ਫਾਰਮੈਟ ਵਿੱਚ ਸੁਰੱਖਿਅਤ ਕੀਤਾ ਜਾਂਦਾ ਹੈ, ਉਦਾਹਰਨ ਲਈ, parquet ਵਿੱਚ, ਜਿਸਨੂੰ ਫਿਰ Impala ਵਿੱਚ ਆਯਾਤ ਕੀਤਾ ਜਾ ਸਕਦਾ ਹੈ। ਇਹ ਉਸ ਸਾਰੇ ਡੇਟਾ ਦੇ ਨਾਲ ਇੱਕ ਟੀਚਾ ਪ੍ਰਦਰਸ਼ਨ ਬਣਾਉਂਦਾ ਹੈ ਜੋ ਇਸ ਬਿੰਦੂ ਤੱਕ ਇਕੱਠੇ ਹੋਏ ਹਨ।

3 ਕਦਮ. ਇੱਕ ਡਾਊਨਲੋਡ ਬਣਾਇਆ ਗਿਆ ਹੈ ਜੋ ਸਟੋਰਫਰੰਟ ਨੂੰ ਹਰ ਰੋਜ਼ ਅੱਪਡੇਟ ਕਰੇਗਾ।
ਵਾਧੇ ਦੀ ਲੋਡਿੰਗ ਦਾ ਸਵਾਲ ਹੈ, ਸ਼ੋਅਕੇਸ ਨੂੰ ਵੰਡਣ ਦੀ ਜ਼ਰੂਰਤ ਹੈ, ਅਤੇ ਸ਼ੋਅਕੇਸ ਦੀ ਆਮ ਸਕੀਮ ਨੂੰ ਬਣਾਈ ਰੱਖਣ ਦਾ ਸਵਾਲ ਹੈ।

ਆਓ ਇੱਕ ਉਦਾਹਰਨ ਲਈਏ। ਦੱਸ ਦੇਈਏ ਕਿ ਇੱਕ ਰਿਪੋਜ਼ਟਰੀ ਬਣਾਉਣ ਦਾ ਪਹਿਲਾ ਕਦਮ ਲਾਗੂ ਕੀਤਾ ਗਿਆ ਹੈ, ਅਤੇ JSON ਫਾਈਲਾਂ ਨੂੰ ਇੱਕ ਫੋਲਡਰ ਵਿੱਚ ਅਪਲੋਡ ਕੀਤਾ ਗਿਆ ਹੈ.

ਉਹਨਾਂ ਤੋਂ ਇੱਕ ਡੇਟਾਫ੍ਰੇਮ ਬਣਾਉਣਾ, ਫਿਰ ਇਸਨੂੰ ਸ਼ੋਅਕੇਸ ਦੇ ਰੂਪ ਵਿੱਚ ਸੁਰੱਖਿਅਤ ਕਰਨਾ, ਕੋਈ ਸਮੱਸਿਆ ਨਹੀਂ ਹੈ. ਇਹ ਸਭ ਤੋਂ ਪਹਿਲਾ ਕਦਮ ਹੈ ਜੋ ਸਪਾਰਕ ਦਸਤਾਵੇਜ਼ਾਂ ਵਿੱਚ ਆਸਾਨੀ ਨਾਲ ਪਾਇਆ ਜਾ ਸਕਦਾ ਹੈ:

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 ਵਾਲਾ ਇੱਕ ਫੋਲਡਰ ਹੈ, ਅਤੇ ਇਸ ਫੋਲਡਰ ਤੋਂ ਇੱਕ ਸ਼ੋਕੇਸ ਬਣਾਇਆ ਗਿਆ ਹੈ। ਸਰੋਤ ਤੋਂ ਡੇਟਾ ਦੇ ਅਗਲੇ ਬੈਚ ਨੂੰ ਲੋਡ ਕਰਨ ਤੋਂ ਬਾਅਦ, ਡੇਟਾ ਮਾਰਟ ਵਿੱਚ ਇੱਕ ਦਿਨ ਦਾ ਡੇਟਾ ਗੁੰਮ ਹੈ।

ਲਾਜ਼ੀਕਲ ਹੱਲ ਸਟੋਰਫਰੰਟ ਨੂੰ ਦਿਨ ਦੇ ਹਿਸਾਬ ਨਾਲ ਵੰਡਣਾ ਹੋਵੇਗਾ, ਜੋ ਹਰ ਅਗਲੇ ਦਿਨ ਇੱਕ ਨਵਾਂ ਭਾਗ ਜੋੜਨ ਦੀ ਇਜਾਜ਼ਤ ਦੇਵੇਗਾ। ਇਸਦੇ ਲਈ ਵਿਧੀ ਵੀ ਚੰਗੀ ਤਰ੍ਹਾਂ ਜਾਣੀ ਜਾਂਦੀ ਹੈ, ਸਪਾਰਕ ਤੁਹਾਨੂੰ ਵੱਖਰੇ ਤੌਰ 'ਤੇ ਭਾਗ ਲਿਖਣ ਦੀ ਆਗਿਆ ਦਿੰਦਾ ਹੈ।

ਪਹਿਲਾਂ, ਅਸੀਂ ਇੱਕ ਸ਼ੁਰੂਆਤੀ ਲੋਡ ਕਰਦੇ ਹਾਂ, ਉੱਪਰ ਦੱਸੇ ਅਨੁਸਾਰ ਡੇਟਾ ਨੂੰ ਸੁਰੱਖਿਅਤ ਕਰਦੇ ਹੋਏ, ਸਿਰਫ਼ ਭਾਗ ਜੋੜਦੇ ਹੋਏ। ਇਸ ਕਾਰਵਾਈ ਨੂੰ ਸਟੋਰਫਰੰਟ ਸ਼ੁਰੂਆਤੀਕਰਣ ਕਿਹਾ ਜਾਂਦਾ ਹੈ ਅਤੇ ਸਿਰਫ ਇੱਕ ਵਾਰ ਕੀਤਾ ਜਾਂਦਾ ਹੈ:

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}

ਮੰਨ ਲਓ ਕਿ ਸਾਡੇ ਕੋਲ ਦੋ ਵੱਖ-ਵੱਖ ਭਾਗ ਹਨ, ਹਰ ਇੱਕ ਲਾਈਨ ਦੇ ਨਾਲ।
ਜਦੋਂ ਅਸੀਂ ਪੂਰੇ ਸਰੋਤ ਡੇਟਾ ਨੂੰ ਪੜ੍ਹਦੇ ਹਾਂ, ਤਾਂ ਸਪਾਰਕ ਕਿਸਮ ਨੂੰ ਨਿਰਧਾਰਤ ਕਰਨ ਦੇ ਯੋਗ ਹੋਵੇਗਾ, ਅਤੇ ਇਹ ਸਮਝੇਗਾ ਕਿ "a" ਕਿਸਮ "ਢਾਂਚਾ" ਦਾ ਇੱਕ ਖੇਤਰ ਹੈ, ਜਿਸ ਵਿੱਚ INT ਕਿਸਮ ਦਾ ਇੱਕ ਨੇਸਟਡ ਖੇਤਰ "b" ਹੈ। ਪਰ, ਜੇਕਰ ਹਰੇਕ ਭਾਗ ਨੂੰ ਵੱਖਰੇ ਤੌਰ 'ਤੇ ਸੁਰੱਖਿਅਤ ਕੀਤਾ ਗਿਆ ਸੀ, ਤਾਂ ਸਾਨੂੰ ਅਸੰਗਤ ਪਾਰਟੀਸ਼ਨ ਸਕੀਮਾਂ ਦੇ ਨਾਲ ਇੱਕ parquet ਮਿਲਦਾ ਹੈ:

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

ਇਹ ਸਥਿਤੀ ਚੰਗੀ ਤਰ੍ਹਾਂ ਜਾਣੀ ਜਾਂਦੀ ਹੈ, ਇਸ ਲਈ ਇੱਕ ਵਿਕਲਪ ਵਿਸ਼ੇਸ਼ ਤੌਰ 'ਤੇ ਜੋੜਿਆ ਗਿਆ ਹੈ - ਸਰੋਤ ਡੇਟਾ ਨੂੰ ਪਾਰਸ ਕਰਦੇ ਸਮੇਂ, ਖਾਲੀ ਖੇਤਰਾਂ ਨੂੰ ਹਟਾਓ:

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

ਇਸ ਸਥਿਤੀ ਵਿੱਚ, ਪਾਰਕਵੇਟ ਵਿੱਚ ਭਾਗ ਹੋਣਗੇ ਜੋ ਇਕੱਠੇ ਪੜ੍ਹੇ ਜਾ ਸਕਦੇ ਹਨ.
ਹਾਲਾਂਕਿ ਜਿਨ੍ਹਾਂ ਨੇ ਅਭਿਆਸ ਵਿੱਚ ਅਜਿਹਾ ਕੀਤਾ ਹੈ ਉਹ ਇੱਥੇ ਕੌੜਾ ਜਿਹਾ ਹੱਸਣਗੇ. ਕਿਉਂ? ਹਾਂ, ਕਿਉਂਕਿ ਦੋ ਹੋਰ ਸਥਿਤੀਆਂ ਹੋਣ ਦੀ ਸੰਭਾਵਨਾ ਹੈ। ਜਾਂ ਤਿੰਨ. ਜਾਂ ਚਾਰ. ਪਹਿਲਾ, ਜੋ ਲਗਭਗ ਨਿਸ਼ਚਿਤ ਤੌਰ 'ਤੇ ਵਾਪਰੇਗਾ, ਇਹ ਹੈ ਕਿ ਵੱਖ-ਵੱਖ JSON ਫਾਈਲਾਂ ਵਿੱਚ ਸੰਖਿਆਤਮਕ ਕਿਸਮਾਂ ਵੱਖਰੀਆਂ ਦਿਖਾਈ ਦੇਣਗੀਆਂ। ਉਦਾਹਰਨ ਲਈ, {intField: 1} ਅਤੇ {intField: 1.1}। ਜੇਕਰ ਅਜਿਹੇ ਖੇਤਰ ਇੱਕ ਭਾਗ ਵਿੱਚ ਪਾਏ ਜਾਂਦੇ ਹਨ, ਤਾਂ ਸਕੀਮਾ ਮਰਜ ਸਭ ਕੁਝ ਸਹੀ ਢੰਗ ਨਾਲ ਪੜ੍ਹੇਗਾ, ਜਿਸ ਨਾਲ ਸਭ ਤੋਂ ਸਹੀ ਕਿਸਮ ਹੋ ਜਾਵੇਗੀ। ਪਰ ਜੇਕਰ ਵੱਖ-ਵੱਖ ਵਿੱਚ, ਤਾਂ ਇੱਕ ਕੋਲ intField: int ਹੋਵੇਗਾ, ਅਤੇ ਦੂਜੇ ਕੋਲ intField: ਡਬਲ ਹੋਵੇਗਾ।

ਇਸ ਸਥਿਤੀ ਨੂੰ ਸੰਭਾਲਣ ਲਈ ਹੇਠ ਲਿਖੇ ਫਲੈਗ ਹਨ:

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

ਹੁਣ ਸਾਡੇ ਕੋਲ ਇੱਕ ਫੋਲਡਰ ਹੈ ਜਿੱਥੇ ਭਾਗ ਹਨ ਜੋ ਇੱਕ ਸਿੰਗਲ ਡੇਟਾਫ੍ਰੇਮ ਵਿੱਚ ਪੜ੍ਹੇ ਜਾ ਸਕਦੇ ਹਨ ਅਤੇ ਪੂਰੇ ਸ਼ੋਕੇਸ ਦੀ ਇੱਕ ਵੈਧ ਪਾਰਕੀਟ ਹੈ। ਹਾਂ? ਨੰ.

ਸਾਨੂੰ ਯਾਦ ਰੱਖਣਾ ਚਾਹੀਦਾ ਹੈ ਕਿ ਅਸੀਂ Hive ਵਿੱਚ ਸਾਰਣੀ ਨੂੰ ਰਜਿਸਟਰ ਕੀਤਾ ਹੈ। Hive ਖੇਤਰ ਦੇ ਨਾਵਾਂ ਵਿੱਚ ਕੇਸ-ਸੰਵੇਦਨਸ਼ੀਲ ਨਹੀਂ ਹੈ, ਜਦੋਂ ਕਿ parquet ਕੇਸ-ਸੰਵੇਦਨਸ਼ੀਲ ਹੈ। ਇਸ ਲਈ, ਸਕੀਮਾਂ ਵਾਲੇ ਭਾਗ: field1: int, ਅਤੇ Field1: int Hive ਲਈ ਇੱਕੋ ਜਿਹੇ ਹਨ, ਪਰ ਸਪਾਰਕ ਲਈ ਨਹੀਂ। ਖੇਤਰ ਦੇ ਨਾਵਾਂ ਨੂੰ ਛੋਟੇ ਅੱਖਰਾਂ ਵਿੱਚ ਬਦਲਣਾ ਨਾ ਭੁੱਲੋ।

ਉਸ ਤੋਂ ਬਾਅਦ, ਸਭ ਕੁਝ ਠੀਕ ਹੋਣ ਲੱਗਦਾ ਹੈ.

ਹਾਲਾਂਕਿ, ਸਾਰੇ ਇੰਨੇ ਸਧਾਰਨ ਨਹੀਂ ਹਨ. ਇੱਕ ਦੂਜੀ, ਜਾਣੀ-ਪਛਾਣੀ ਸਮੱਸਿਆ ਵੀ ਹੈ। ਕਿਉਂਕਿ ਹਰੇਕ ਨਵਾਂ ਭਾਗ ਵੱਖਰੇ ਤੌਰ 'ਤੇ ਸੁਰੱਖਿਅਤ ਕੀਤਾ ਜਾਂਦਾ ਹੈ, ਭਾਗ ਫੋਲਡਰ ਵਿੱਚ ਸਪਾਰਕ ਸਰਵਿਸ ਫਾਈਲਾਂ ਸ਼ਾਮਲ ਹੋਣਗੀਆਂ, ਉਦਾਹਰਨ ਲਈ, _SUCCESS ਓਪਰੇਸ਼ਨ ਸਫਲਤਾ ਫਲੈਗ। ਇਸ ਦੇ ਨਤੀਜੇ ਵਜੋਂ ਪੈਰਕੇਟ ਕਰਨ ਦੀ ਕੋਸ਼ਿਸ਼ ਕਰਦੇ ਸਮੇਂ ਇੱਕ ਗਲਤੀ ਹੋਵੇਗੀ। ਇਸ ਤੋਂ ਬਚਣ ਲਈ, ਤੁਹਾਨੂੰ ਸਪਾਰਕ ਨੂੰ ਫੋਲਡਰ ਵਿੱਚ ਸੇਵਾ ਫਾਈਲਾਂ ਜੋੜਨ ਤੋਂ ਰੋਕਣ ਲਈ ਸੰਰਚਨਾ ਨੂੰ ਕੌਂਫਿਗਰ ਕਰਨ ਦੀ ਲੋੜ ਹੈ:

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

ਅਜਿਹਾ ਲਗਦਾ ਹੈ ਕਿ ਹੁਣ ਹਰ ਰੋਜ਼ ਇੱਕ ਨਵਾਂ ਪਾਰਕੁਏਟ ਭਾਗ ਟਾਰਗਿਟ ਸ਼ੋਕੇਸ ਫੋਲਡਰ ਵਿੱਚ ਜੋੜਿਆ ਜਾਂਦਾ ਹੈ, ਜਿੱਥੇ ਦਿਨ ਲਈ ਪਾਰਸ ਕੀਤਾ ਡੇਟਾ ਸਥਿਤ ਹੁੰਦਾ ਹੈ। ਅਸੀਂ ਪਹਿਲਾਂ ਹੀ ਧਿਆਨ ਰੱਖਿਆ ਸੀ ਕਿ ਡੇਟਾ ਕਿਸਮ ਦੇ ਅਪਵਾਦ ਵਾਲੇ ਕੋਈ ਭਾਗ ਨਹੀਂ ਹਨ।

ਪਰ, ਸਾਡੇ ਕੋਲ ਇੱਕ ਤੀਜੀ ਸਮੱਸਿਆ ਹੈ. ਹੁਣ ਆਮ ਸਕੀਮਾ ਦਾ ਪਤਾ ਨਹੀਂ ਹੈ, ਇਸ ਤੋਂ ਇਲਾਵਾ, Hive ਵਿੱਚ ਸਾਰਣੀ ਵਿੱਚ ਇੱਕ ਗਲਤ ਸਕੀਮਾ ਹੈ, ਕਿਉਂਕਿ ਹਰੇਕ ਨਵੇਂ ਭਾਗ ਨੇ ਸਕੀਮਾ ਵਿੱਚ ਇੱਕ ਵਿਗਾੜ ਪੇਸ਼ ਕੀਤਾ ਹੈ।

ਤੁਹਾਨੂੰ ਟੇਬਲ ਨੂੰ ਦੁਬਾਰਾ ਰਜਿਸਟਰ ਕਰਨ ਦੀ ਲੋੜ ਹੈ। ਇਹ ਆਸਾਨੀ ਨਾਲ ਕੀਤਾ ਜਾ ਸਕਦਾ ਹੈ: ਸਟੋਰਫਰੰਟ ਦੇ ਪੈਰਕੇਟ ਨੂੰ ਦੁਬਾਰਾ ਪੜ੍ਹੋ, ਸਕੀਮਾ ਲਓ ਅਤੇ ਇਸਦੇ ਆਧਾਰ 'ਤੇ ਇੱਕ DDL ਬਣਾਓ, ਜਿਸ ਨਾਲ Hive ਵਿੱਚ ਫੋਲਡਰ ਨੂੰ ਇੱਕ ਬਾਹਰੀ ਸਾਰਣੀ ਦੇ ਤੌਰ 'ਤੇ ਦੁਬਾਰਾ ਰਜਿਸਟਰ ਕਰਨਾ ਹੈ, ਟਾਰਗੇਟ ਸਟੋਰਫਰੰਟ ਦੀ ਸਕੀਮਾ ਨੂੰ ਅੱਪਡੇਟ ਕਰਨਾ।

ਸਾਡੇ ਕੋਲ ਚੌਥੀ ਸਮੱਸਿਆ ਹੈ। ਜਦੋਂ ਅਸੀਂ ਪਹਿਲੀ ਵਾਰ ਟੇਬਲ ਨੂੰ ਰਜਿਸਟਰ ਕੀਤਾ, ਅਸੀਂ ਸਪਾਰਕ 'ਤੇ ਭਰੋਸਾ ਕੀਤਾ। ਹੁਣ ਅਸੀਂ ਇਹ ਆਪਣੇ ਆਪ ਕਰਦੇ ਹਾਂ, ਅਤੇ ਸਾਨੂੰ ਇਹ ਯਾਦ ਰੱਖਣ ਦੀ ਜ਼ਰੂਰਤ ਹੈ ਕਿ ਪਾਰਕਵੇਟ ਖੇਤਰ ਉਹਨਾਂ ਪਾਤਰਾਂ ਨਾਲ ਸ਼ੁਰੂ ਹੋ ਸਕਦੇ ਹਨ ਜਿਨ੍ਹਾਂ ਦੀ Hive ਲਈ ਆਗਿਆ ਨਹੀਂ ਹੈ. ਉਦਾਹਰਨ ਲਈ, ਸਪਾਰਕ ਲਾਈਨਾਂ ਨੂੰ ਬਾਹਰ ਸੁੱਟਦਾ ਹੈ ਜੋ ਇਹ "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(",", ",`") ਬਦਲੋ("ਐਰੇ<`", "ਐਰੇ<") ਸੁਰੱਖਿਅਤ DDL ਬਣਾਉਂਦਾ ਹੈ, ਜਿਵੇਂ ਕਿ:

create table tname (_field1 string, 1field string)

"_field1, 1field" ਵਰਗੇ ਫੀਲਡ ਨਾਵਾਂ ਦੇ ਨਾਲ, ਸੁਰੱਖਿਅਤ DDL ਬਣਾਇਆ ਜਾਂਦਾ ਹੈ ਜਿੱਥੇ ਫੀਲਡ ਨਾਮ ਬਚੇ ਹੋਏ ਹਨ: ਟੇਬਲ ਬਣਾਓ `tname` (`_field1` ਸਤਰ, `1field` ਸਤਰ)।

ਸਵਾਲ ਪੈਦਾ ਹੁੰਦਾ ਹੈ: ਪੂਰੀ ਸਕੀਮਾ (ਪੀਐਫ ਕੋਡ ਵਿੱਚ) ਦੇ ਨਾਲ ਇੱਕ ਡੇਟਾਫ੍ਰੇਮ ਨੂੰ ਸਹੀ ਢੰਗ ਨਾਲ ਕਿਵੇਂ ਪ੍ਰਾਪਤ ਕਰਨਾ ਹੈ? ਇਹ ਪੀਐਫ ਕਿਵੇਂ ਪ੍ਰਾਪਤ ਕੀਤਾ ਜਾਵੇ? ਇਹ ਪੰਜਵੀਂ ਸਮੱਸਿਆ ਹੈ। ਟਾਰਗੇਟ ਸ਼ੋਕੇਸ ਦੀਆਂ ਪਾਰਕੁਏਟ ਫਾਈਲਾਂ ਵਾਲੇ ਫੋਲਡਰ ਤੋਂ ਸਾਰੇ ਭਾਗਾਂ ਦੀ ਸਕੀਮ ਨੂੰ ਮੁੜ ਪੜ੍ਹੋ? ਇਹ ਤਰੀਕਾ ਸਭ ਤੋਂ ਸੁਰੱਖਿਅਤ ਹੈ, ਪਰ ਮੁਸ਼ਕਲ ਹੈ.

ਸਕੀਮਾ ਪਹਿਲਾਂ ਹੀ Hive ਵਿੱਚ ਹੈ। ਤੁਸੀਂ ਪੂਰੀ ਟੇਬਲ ਦੀ ਸਕੀਮਾ ਅਤੇ ਨਵੇਂ ਭਾਗ ਨੂੰ ਮਿਲਾ ਕੇ ਇੱਕ ਨਵੀਂ ਸਕੀਮਾ ਪ੍ਰਾਪਤ ਕਰ ਸਕਦੇ ਹੋ। ਇਸ ਲਈ ਤੁਹਾਨੂੰ Hive ਤੋਂ ਟੇਬਲ ਸਕੀਮਾ ਲੈਣ ਅਤੇ ਇਸਨੂੰ ਨਵੇਂ ਭਾਗ ਦੀ ਸਕੀਮਾ ਨਾਲ ਜੋੜਨ ਦੀ ਲੋੜ ਹੈ। ਇਹ Hive ਤੋਂ ਟੈਸਟ ਮੈਟਾਡੇਟਾ ਨੂੰ ਪੜ੍ਹ ਕੇ, ਇਸਨੂੰ ਇੱਕ ਅਸਥਾਈ ਫੋਲਡਰ ਵਿੱਚ ਸੁਰੱਖਿਅਤ ਕਰਕੇ, ਅਤੇ ਸਪਾਰਕ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਦੋਵੇਂ ਭਾਗਾਂ ਨੂੰ ਇੱਕੋ ਵਾਰ ਪੜ੍ਹ ਕੇ ਕੀਤਾ ਜਾ ਸਕਦਾ ਹੈ।

ਅਸਲ ਵਿੱਚ, ਇੱਥੇ ਸਭ ਕੁਝ ਹੈ ਜਿਸਦੀ ਤੁਹਾਨੂੰ ਲੋੜ ਹੈ: 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 ਨੂੰ ਪੜ੍ਹਨ ਅਤੇ ਇਸ ਦੇ ਆਧਾਰ 'ਤੇ ਸਟੋਰਫਰੰਟ ਬਣਾਉਣ ਦਾ ਸਧਾਰਨ ਕੰਮ ਬਹੁਤ ਸਾਰੀਆਂ ਅਨਿੱਖੜਵਾਂ ਮੁਸ਼ਕਲਾਂ ਨੂੰ ਦੂਰ ਕਰਦਾ ਹੈ, ਹੱਲ ਜਿਨ੍ਹਾਂ ਲਈ ਤੁਹਾਨੂੰ ਵੱਖਰੇ ਤੌਰ 'ਤੇ ਖੋਜ ਕਰਨੀ ਪਵੇਗੀ। ਅਤੇ ਹਾਲਾਂਕਿ ਇਹ ਹੱਲ ਸਧਾਰਨ ਹਨ, ਇਹਨਾਂ ਨੂੰ ਲੱਭਣ ਵਿੱਚ ਬਹੁਤ ਸਮਾਂ ਲੱਗਦਾ ਹੈ।

ਸ਼ੋਅਕੇਸ ਦੀ ਉਸਾਰੀ ਨੂੰ ਲਾਗੂ ਕਰਨ ਲਈ, ਮੈਨੂੰ ਇਹ ਕਰਨਾ ਪਿਆ:

• ਸ਼ੋਕੇਸ ਵਿੱਚ ਭਾਗ ਜੋੜੋ, ਸੇਵਾ ਫਾਈਲਾਂ ਤੋਂ ਛੁਟਕਾਰਾ ਪਾਓ
• ਸਪਾਰਕ ਦੁਆਰਾ ਟਾਈਪ ਕੀਤੇ ਸਰੋਤ ਡੇਟਾ ਵਿੱਚ ਖਾਲੀ ਖੇਤਰਾਂ ਨਾਲ ਨਜਿੱਠੋ
• ਸਧਾਰਨ ਕਿਸਮਾਂ ਨੂੰ ਇੱਕ ਸਤਰ ਵਿੱਚ ਕਾਸਟ ਕਰੋ
• ਖੇਤਰ ਦੇ ਨਾਵਾਂ ਨੂੰ ਛੋਟੇ ਅੱਖਰਾਂ ਵਿੱਚ ਬਦਲੋ
• Hive (DDL ਜਨਰੇਸ਼ਨ) ਵਿੱਚ ਵੱਖਰਾ ਡਾਟਾ ਅੱਪਲੋਡ ਅਤੇ ਟੇਬਲ ਰਜਿਸਟ੍ਰੇਸ਼ਨ
• ਖੇਤਰ ਦੇ ਨਾਵਾਂ ਤੋਂ ਬਚਣਾ ਨਾ ਭੁੱਲੋ ਜੋ Hive ਨਾਲ ਅਸੰਗਤ ਹੋ ਸਕਦੇ ਹਨ
• Hive ਵਿੱਚ ਟੇਬਲ ਰਜਿਸਟ੍ਰੇਸ਼ਨ ਨੂੰ ਕਿਵੇਂ ਅੱਪਡੇਟ ਕਰਨਾ ਹੈ ਬਾਰੇ ਜਾਣੋ

ਸੰਖੇਪ ਵਿੱਚ, ਅਸੀਂ ਨੋਟ ਕਰਦੇ ਹਾਂ ਕਿ ਦੁਕਾਨ ਦੀਆਂ ਖਿੜਕੀਆਂ ਬਣਾਉਣ ਦਾ ਫੈਸਲਾ ਬਹੁਤ ਸਾਰੀਆਂ ਕਮੀਆਂ ਨਾਲ ਭਰਿਆ ਹੋਇਆ ਹੈ। ਇਸ ਲਈ, ਲਾਗੂ ਕਰਨ ਵਿੱਚ ਮੁਸ਼ਕਲਾਂ ਦੇ ਮਾਮਲੇ ਵਿੱਚ, ਸਫਲ ਮੁਹਾਰਤ ਵਾਲੇ ਇੱਕ ਤਜਰਬੇਕਾਰ ਸਾਥੀ ਨਾਲ ਸੰਪਰਕ ਕਰਨਾ ਬਿਹਤਰ ਹੈ.

ਇਸ ਲੇਖ ਨੂੰ ਪੜ੍ਹਨ ਲਈ ਤੁਹਾਡਾ ਧੰਨਵਾਦ, ਅਸੀਂ ਉਮੀਦ ਕਰਦੇ ਹਾਂ ਕਿ ਤੁਹਾਨੂੰ ਜਾਣਕਾਰੀ ਲਾਭਦਾਇਕ ਲੱਗੇਗੀ।

ਸਰੋਤ: www.habr.com