استيعاب البيانات كنوع متغير شبه منظم

هام

هذه الميزة في المعاينة العامة.

في Databricks Runtime 15.3 وما فوق، يمكنك استخدام VARIANT النوع لاستيعاب البيانات شبه المنظمة. توضح هذه المقالة السلوك وتوفر أنماطا أمثلة لاستيعاب البيانات من تخزين الكائنات السحابية باستخدام أداة التحميل التلقائي وأوامر COPY INTOتدفق السجلات من Kafka وأوامر SQL لإنشاء جداول جديدة باستخدام بيانات متغيرة أو إدراج سجلات جديدة باستخدام نوع المتغير.

راجع بيانات متغير الاستعلام.

إنشاء جدول بعمود متغير

VARIANT هو نوع SQL قياسي في Databricks Runtime 15.3 وما فوق ويدعمه جداول مدعومة من Delta Lake. تستخدم الجداول المدارة على Azure Databricks Delta Lake بشكل افتراضي، بحيث يمكنك إنشاء جدول فارغ بعمود واحد VARIANT باستخدام بناء الجملة التالي:

CREATE TABLE table_name (variant_column VARIANT)

بدلا من ذلك، يمكنك استخدام الدالة PARSE_JSON على سلسلة JSON لاستخدام عبارة CTAS لإنشاء جدول بعمود متغير. ينشئ المثال التالي جدولا بعمودين:

• id العمود المستخرج من سلسلة JSON كنوعSTRING.
• variant_column يحتوي العمود على سلسلة JSON بأكملها مرمزة كنوعVARIANT.

CREATE TABLE table_name AS
  SELECT json_string:id AS id,
    PARSE_JSON(json_string) variant_column
  FROM source_data

إشعار

VARIANT لا يمكن استخدام الأعمدة لتكتل المفاتيح أو الأقسام أو مفاتيح ترتيب Z. لا يمكن استخدام البيانات المخزنة بنوع VARIANT للمقارنات وترتيبها.

توصي Databricks باستخراج الحقول وتخزينها كأعمدة غير متغيرة تخطط لاستخدامها لتسريع الاستعلامات وتحسين تخطيط التخزين.

إدراج بيانات باستخدام parse_json

إذا كان الجدول الهدف يحتوي بالفعل على عمود مرمز ك VARIANT، يمكنك استخدام parse_json لإدراج سجلات سلسلة JSON ك VARIANT، كما في المثال التالي:

SQL

INSERT INTO table_name (variant_column)
  SELECT PARSE_JSON(json_string)
  FROM source_data

Python

from pyspark.sql.functions import col, parse_json

(spark.read
  .table("source_data")
  .select(parse_json(col("json_string")))
  .write
  .mode("append")
  .saveAsTable("table_name")
)

استيعاب البيانات من تخزين كائن السحابة كمتغير

في Databricks Runtime 15.3 وما فوق، يمكنك استخدام Auto Loader لتحميل جميع البيانات من مصادر JSON ك عمود واحد VARIANT في جدول هدف. نظرا VARIANT لأن هذا النمط مرن في المخطط وتغييرات النوع ويحافظ على حساسية NULL الحالة والقيم الموجودة في مصدر البيانات، فإن هذا النمط قوي لمعظم سيناريوهات الاستيعاب مع المحاذير التالية:

• لا يمكن ترميز سجلات JSON التي تم تكوينها بشكل غير صحيح باستخدام VARIANT النوع .
• VARIANT يمكن أن يحتوي النوع على سجلات يصل حجمها إلى 16 ميغابايت فقط.

إشعار

يعامل المتغير سجلات كبيرة جدا مشابهة للسجلات التالفة. في وضع المعالجة الافتراضي PERMISSIVE ، يتم التقاط سجلات كبيرة جدا في _malformed_data العمود جنبا إلى جنب مع سجلات JSON التي تم تكوينها بشكل غير جيد.

نظرا لأنه يتم تسجيل جميع البيانات من مصدر JSON كعمود واحد VARIANT ، لا يحدث تطور مخطط أثناء الاستيعاب ولا rescuedDataColumn يتم دعمه. يفترض المثال التالي أن الجدول الهدف موجود بالفعل بعمود واحد VARIANT .

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("singleVariantColumn", "variant_column")
  .load("/Volumes/catalog_name/schema_name/volume_name/path")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("table_name")
)

يمكنك أيضا تحديد VARIANT عند تعريف مخطط أو تمرير schemaHints. يجب أن تحتوي البيانات الموجودة في حقل المصدر المشار إليه على سلسلة JSON صالحة. توضح الأمثلة التالية بناء الجملة هذا:

# Define the schema.
# Writes the columns `name` as a string and `address` as variant.
(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .schema("name STRING, address VARIANT")
  .load("/Volumes/catalog_name/schema_name/volume_name/path")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("table_name")
)

# Define the schema.
# A single field `payload` containing JSON data is written as variant.
(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .schema("payload VARIANT")
  .load("/Volumes/catalog_name/schema_name/volume_name/path")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("table_name")
)

# Supply schema hints.
# Writes the `address` column as variant.
# Infers the schema for other fields using standard rules.
(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "address VARIANT")
  .load("/Volumes/catalog_name/schema_name/volume_name/path")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("table_name")
)

استخدام COPY INTO مع متغير

توصي Databricks باستخدام أداة COPY INTO التحميل التلقائي عند توفرها.

COPY INTO يدعم استيعاب محتويات مصدر بيانات JSON بالكامل كعمود واحد. ينشئ المثال التالي جدولا جديدا بعمود واحد VARIANT ثم يستخدم COPY INTO لاستيعاب السجلات من مصدر ملف JSON.

CREATE TABLE table_name (variant_column VARIANT);
COPY INTO table_name
  FROM '/Volumes/catalog_name/schema_name/volume_name/path'
  FILEFORMAT = JSON
  FORMAT_OPTIONS ('singleVariantColumn' = 'name')

يمكنك أيضا تعريف أي حقل في الجدول الهدف على أنه VARIANT. عند تشغيل COPY INTO، يتم استيعاب الحقول المقابلة في مصدر البيانات وإرسالها للكتابة VARIANT ، كما في الأمثلة التالية:

-- Extracts the `address` field from the JSON record and casts to variant
CREATE TABLE table_name (address VARIANT);
COPY INTO table_name
  FROM '/Volumes/catalog_name/schema_name/volume_name/path'
  FILEFORMAT = JSON

-- Extracts `name` and `address` from the JSON record and casts `address` to variant
CREATE TABLE table_name (name STRING, address VARIANT);
COPY INTO table_name
  FROM '/Volumes/catalog_name/schema_name/volume_name/path'
  FILEFORMAT = JSON

دفق بيانات Kafka كمتغير

العديد من تدفقات Kafka ترميز حمولاتها باستخدام JSON. إن استيعاب تدفقات Kafka باستخدام VARIANT يجعل أحمال العمل هذه قوية لتغييرات المخطط.

يوضح المثال التالي قراءة مصدر دفق Kafka، وإرسال key ك STRING و value ك VARIANT، والكتابة إلى جدول هدف.

from pyspark.sql.functions import col, parse_json

(spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .option("startingOffsets", "earliest")
  .load()
  .select(
    col("key").cast("string"),
    parse_json(col("value").cast("string"))
  ).writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("table_name")
)