MaxCompute:Tabel eksternal Parquet

Sintaks

Saat skema file Parquet berbeda dengan skema tabel eksternal:

• Ketidaksesuaian jumlah kolom: Jika file Parquet memiliki lebih sedikit kolom daripada yang didefinisikan dalam DDL tabel eksternal, kolom yang hilang mengembalikan NULL. Jika file memiliki lebih banyak kolom, kolom tambahan diabaikan.

• Ketidaksesuaian tipe kolom: Jika tipe kolom dalam file Parquet tidak sesuai dengan tipe yang sesuai dalam DDL, operasi baca gagal. Misalnya, error seperti ODPS-0123131:User defined function exception - Traceback:xxx dilaporkan jika Anda mencoba membaca kolom INT sebagai bidang STRING.

CREATE EXTERNAL TABLE [IF NOT EXISTS] <mc_oss_extable_name>
(
  <col_name> <data_type>,
  ...
)
[COMMENT <table_comment>]
[PARTITIONED BY (<col_name> <data_type>, ...)]
STORED AS parquet 
LOCATION '<oss_location>' 
[tblproperties ('<tbproperty_name>'='<tbproperty_value>',...)];

CREATE EXTERNAL TABLE [IF NOT EXISTS] <mc_oss_extable_name>
(
  <col_name> <data_type>,
  ...
)
[COMMENT <table_comment>]
[PARTITIONED BY (<col_name> <data_type>, ...)]
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
WITH serdeproperties(
    'odps.properties.rolearn'='acs:ram::<uid>:role/<role_name>',
    'mcfed.parquet.compression'='ZSTD/SNAPPY/GZIP'
)
STORED AS parquet 
LOCATION '<oss_location>' 
;

Parameter umum

Untuk informasi selengkapnya tentang parameter umum, lihat Parameter sintaks dasar.

Parameter unik

Daftar putih dan blacklist

Tabel eksternal OSS MaxCompute mendukung penyaringan daftar putih dan blacklist. Dengan mengatur parameter daftar putih dan blacklist dalam tblproperties, Anda dapat menyaring file mana yang akan dibaca dari direktori. Untuk detailnya, lihat Daftar putih dan blacklist.

Penggunaan

• Aktifkan Predicate Push Down (PPD)

  Sebelum menjalankan kueri SQL, gunakan perintah set untuk mengatur dua parameter tingkat sesi berikut guna mengaktifkan Parquet PPD.

  -- Aktifkan pembaca native Parquet. 
  set odps.ext.parquet.native = true; 
  -- Aktifkan Parquet PPD.
  set odps.sql.parquet.use.predicate.pushdown = true; 

• Contoh

  Contoh ini menggunakan set data uji TPC-DS 1 TB dan tabel eksternal Parquet tpcds_1t_store_sales. Dalam contoh ini, PPD diaktifkan dan kueri filter dijalankan. Volume data total adalah 2.879.987.999 baris.

  -- Buat tabel eksternal tpcds_1t_store_sales.
  CREATE EXTERNAL TABLE IF NOT EXISTS tpcds_1t_store_sales (
      ss_sold_date_sk         BIGINT,
      ss_sold_time_sk         BIGINT,
      ss_item_sk              BIGINT,
      ss_customer_sk          BIGINT,
      ss_cdemo_sk             BIGINT,
      ss_hdemo_sk             BIGINT,
      ss_addr_sk              BIGINT,
      ss_store_sk             BIGINT,
      ss_promo_sk             BIGINT,
      ss_ticket_number        BIGINT,
      ss_quantity             BIGINT,
      ss_wholesale_cost       DECIMAL(7,2),
      ss_list_price           DECIMAL(7,2),
      ss_sales_price          DECIMAL(7,2),
      ss_ext_discount_amt     DECIMAL(7,2),
      ss_ext_sales_price      DECIMAL(7,2),
      ss_ext_wholesale_cost   DECIMAL(7,2),
      ss_ext_list_price       DECIMAL(7,2),
      ss_ext_tax              DECIMAL(7,2),
      ss_coupon_amt           DECIMAL(7,2),
      ss_net_paid             DECIMAL(7,2),
      ss_net_paid_inc_tax     DECIMAL(7,2),
      ss_net_profit           DECIMAL(7,2)
  )
  ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
  WITH serdeproperties(
    'odps.properties.rolearn'='acs:ram::<uid>:role/<role_name>',
    'mcfed.parquet.compression'='zstd'
  )
  STORED AS parquet
  LOCATION 'oss://oss-cn-hangzhou-internal.aliyuncs.com/oss_bucket_path/';
  -- Gunakan set data uji TPC-DS 1 TB.
  INSERT OVERWRITE TABLE tpcds_1t_store_sales
  SELECT 
      ss_sold_date_sk,
      ss_sold_time_sk,
      ss_item_sk,
      ss_customer_sk,
      ss_cdemo_sk,
      ss_hdemo_sk,
      ss_addr_sk,
      ss_store_sk,
      ss_promo_sk,
      ss_ticket_number,
      ss_quantity,
      ss_wholesale_cost,
      ss_list_price,
      ss_sales_price,
      ss_ext_discount_amt,
      ss_ext_sales_price,
      ss_ext_wholesale_cost,
      ss_ext_list_price,
      ss_ext_tax,
      ss_coupon_amt,
      ss_net_paid,
      ss_net_paid_inc_tax,
      ss_net_profit
  FROM 
      bigdata_public_dataset.tpcds_1t.store_sales;
  -- Jalankan kueri.
  SELECT SUM(ss_sold_date_sk) FROM tpcds_1t_store_sales
    WHERE ss_sold_date_sk >= 2451871 AND ss_sold_date_sk <= 2451880;

Perbandingan performa

Mengaktifkan PPD mengurangi jumlah data yang dipindai, sehingga menurunkan latensi kueri dan konsumsi resource.

Detail pengujian

1. Tabel eksternal Parquet tanpa PPD

   SET odps.ext.parquet.native = true;
   SET odps.sql.parquet.use.predicate.pushdown = false;
   SELECT SUM(ss_sold_date_sk) FROM tpcds_1t_store_sales
     WHERE ss_store_sk = 2 AND ss_sold_date_sk >= 2451871 AND ss_sold_date_sk <= 2451880;

   

   Hasil menunjukkan bahwa untuk tugas M1 dalam Fuxi Jobs, IO Records Input adalah 2,9 G, IO Bytes Input adalah 18,06 GB, dan Latency adalah 00:00:18,000.

   Tab Summary hasil eksekusi menunjukkan bahwa konsumsi resource adalah cpu 19,25 Core × Min dan memori 24,07 GB × Min. Waktu proses job adalah 23,000 detik, dan mode proses adalah fuxi job 2.0. Tugas M1 memiliki 1.404 instans, waktu proses 18,000 detik, 2.879.987.999 record input, dan 355 record output. Tugas R2_1 memiliki 1 instans, waktu proses 4,000 detik, dan 1 record output.

2. Tabel eksternal Parquet dengan PPD

   SET odps.ext.parquet.native = true;
   SET odps.sql.parquet.use.predicate.pushdown = true;
   SELECT SUM(ss_sold_date_sk) FROM tpcds_1t_store_sales
     WHERE ss_store_sk = 2 AND ss_sold_date_sk >= 2451871 AND ss_sold_date_sk <= 2451880;

   

   Setelah menjalankan kueri ini, ringkasan job menunjukkan bahwa konsumsi resource adalah cpu 11,47 Core × Min, memory 14,33 GB × Min, dengan total waktu proses 15 detik. Tahap M1 memiliki 1.404 instans, waktu proses 12 detik, dan 762.366.649 record input (sekitar 3.339.386.880 byte). Tahap R2_1 memiliki 1 instans dan waktu proses 3 detik.

   Banyak mapper kosong dan tidak perlu membaca data:

   

   Log pemangkasan grup baris aktual:

   [2024-05-10 22:29:22.692182]    [INFO]   [239551]    [/home/admin/odps_build/workspace/IRDS_CMK_7u/jenkins-IRDS_CMK_7u-70
   16/common/table/file_formats/parquet/parquet_row_group_pruner.cpp:100]    The expression to prune row groups:(((ss_store_s
   k == 2:int64) and (ss_sold_date_sk >= 2451871:int64)) and (ss_sold_date_sk <= 2451880:int64))
   [2024-05-10 22:29:22.705508]    [INFO]   [239551]    [/home/admin/odps_build/workspace/IRDS_CMK_7u/jenkins-IRDS_CMK_7u-70
   16/common/table/file_formats/parquet/parquet_reader_factory.cpp:136]    Parquet row group pruning is enabled, millisecon
   ds elapsed:13    Total row group count:1 Pruned row group count:1    The first several row group indexes:
   [2024-05-10 22:29:22.705532]    [INFO]   [239551]    [/home/admin/odps_build/workspace/IRDS_CMK_7u/jenkins-IRDS_CMK_7u-70
   16/common/table/file_formats/parquet/parquet_reader_factory.cpp:60]  total feasible parquet row group count:0]

3. Tabel internal dengan PPD

   Efek pemangkasan lebih signifikan karena data dalam tabel internal diurutkan.

   SELECT SUM(ss_sold_date_sk) FROM bigdata_public_dataset.tpcds_1t.store_sales
     WHERE ss_store_sk = 2 AND ss_sold_date_sk >= 2451871 AND ss_sold_date_sk <= 2451880;

   Setelah menjalankan kueri ini, DAG job menunjukkan bahwa jumlah total baris dalam sumber data adalah 2.879.987.999, tetapi jumlah baris yang benar-benar dipindai hanya 32.830.000. Tahap M1 (703 instans) membaca 32.830.000 baris dan menghasilkan 323 baris. Tahap R2_1 menerima 323 baris dan menghasilkan 1 baris. Hal ini menunjukkan bahwa efek pemangkasan sangat signifikan saat PPD diaktifkan untuk tabel internal dengan data terurut.

   Pemantauan Fuxi Jobs menunjukkan bahwa job SQL_0_1_0_job_0 selesai. Job ini mencakup dua Tugas Fuxi, M1 dan R2_1, keduanya dengan status Terminated. M1 memiliki 703 instans, input 32,8 juta record (1,52 GB), output 323 record, dan latensi 00:00:09,375. R2_1 memiliki 1 instans, input 323 record, output 1 record, dan latensi 00:00:03,873. Detail instans M1 menunjukkan 4 instans Data-Skew. Instans seperti M1#101_0, M1#103_0, dan M1#105_0 memiliki 0 untuk Input dan Output. Hal ini menunjukkan bahwa instans tersebut adalah instans dry-run dan kueri ini memiliki masalah kesenjangan data.

   resource cost: cpu 5,62 Core * Min, memory 7,02 GB * Min
   inputs:
       lakehouse47_2.default.tpcds_1t_store_sales2: 32830000 (1633880386 bytes)
   outputs:
   Job run time: 14,000
   Job run mode: fuxi job 2.0
   Job run engine: execution engine
   M1:
       instance count: 703
       run time: 9,000
       instance time:
           min: 0,000, max: 2,000, avg: 0,000
       input records:
           TableScan1: 32830000  (min: 0, max: 210000, avg: 46699)
       output records:
           StreamLineWrite1: 323  (min: 0, max: 1, avg: 0)
       metrics_output_count:
           Calc1: 58025  (min: 0, max: 461, avg: 82)
           HashAgg1: 323  (min: 0, max: 1, avg: 0)
           StreamLineWrite1: 323  (min: 0, max: 1, avg: 0)
           TableScan1: 32830000  (min: 0, max: 210000, avg: 46699)
       metrics_inner_time_ms:
           Calc1: 4  (min: 0, max: 2, avg: 0)  MaxInstance: 21
           GlobalInit: 57752  (min: 60, max: 386, avg: 82)  MaxInstance: 17
           HashAgg1: 0  (min: 0, max: 0, avg: 0)   MaxInstance: 2
           StreamLineWrite1: 20469  (min: 5, max: 899, avg: 29)   MaxInstance: 400
           TableScan1: 131977  (min: 51, max: 1417, avg: 187)  MaxInstance: 301
   R2_1:
       instance count: 1
       run time: 4,000
       instance time:
           min: 0,000, max: 0,000, avg: 0,000
       input records:
           StreamLineRead1: 323  (min: 323, max: 323, avg: 323)
       output records:
           AdhocSink1: 1  (min: 1, max: 1, avg: 1)
       metrics_output_count:
           AdhocSink1: 1  (min: 1, max: 1, avg: 1)

Performa kueri

Set data: TPC-DS 1 TB

Resource: 900+ CU

Metode pengujian: ETL

• Latensi: ZSTD 13,4% lebih cepat daripada Snappy dan 16,5% lebih cepat daripada tidak terkompresi.

• CPU: ZSTD menggunakan 40,7% lebih sedikit CPU daripada Snappy dan 57,75% lebih sedikit daripada tidak terkompresi.

• Memori: ZSTD menggunakan 39,04% lebih sedikit memori daripada Snappy dan 56,13% lebih sedikit daripada tidak terkompresi.

• Storage: ZSTD menggunakan 31,15% lebih sedikit ruang penyimpanan daripada Snappy dan 45,8% lebih sedikit daripada tidak terkompresi.

Efisiensi penyimpanan

Set data: TPC-DS 1 TB

Resource: 900+ CU

Metode pengujian: ETL

• tidak terkompresi: Meskipun format ini tidak dikompresi, menghasilkan volume data yang lebih besar dan overhead I/O yang lebih tinggi, yang menyebabkan performa keseluruhan buruk.

• snappy: Kecepatan kompresi tidak lebih cepat daripada ZSTD tingkat rendah, tetapi rasio kompresinya lebih tinggi. Hal ini menghasilkan performa keseluruhan lebih buruk daripada ZSTD tingkat rendah.

• zstd: Ukuran data output konvergen dengan cepat. Tingkat yang lebih tinggi memberikan kompresi tambahan minimal (hanya 13,87% lebih banyak) tetapi menyebabkan peningkatan cepat dalam waktu dan konsumsi resource, yang secara drastis mengurangi efektivitas biaya. Untuk skenario ini, ZSTD tingkat rendah (level 3 hingga 5) memberikan hasil terbaik. Level 3 adalah default.

• Pada set data TPC-DS 1 TB, ZSTD menggunakan 31,1% lebih sedikit ruang penyimpanan daripada Snappy dan 45,8% lebih sedikit daripada tidak terkompresi.

Ketidaksesuaian tipe kolom antara file Parquet dan DDL tabel eksternal

• Pesan error

  ODPS-0123131:User defined function exception - Traceback:
  java.lang.ClassCastException: org.apache.hadoop.io.LongWritable cannot be cast to org.apache.hadoop.io.IntWritable 
     at org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableIntObjectInspector.getPrimitiveJavaObject(WritableIntObjectInspector.java:46)

• Deskripsi error

  Tipe bidang LongWritable dari file Parquet tidak sesuai dengan tipe INT dalam DDL tabel eksternal.

• Solusi

  Ubah tipe INT dalam DDL tabel eksternal menjadi BIGINT.

Error saat menulis ke tabel eksternal: java.lang.OutOfMemoryError

• Pesan error

  ODPS-0123131:User defined function exception - Traceback:
  java.lang.OutOfMemoryError: Java heap space
  	at java.io.ByteArrayOutputStream.<init>(ByteArrayOutputStream.java:77)
  	at org.apache.parquet.bytes.BytesInput$BAOS.<init>(BytesInput.java:175)
  	at org.apache.parquet.bytes.BytesInput$BAOS.<init>(BytesInput.java:173)
  	at org.apache.parquet.bytes.BytesInput.toByteArray(BytesInput.java:161)

• Deskripsi error

  Terjadi error kehabisan memori (OOM) saat Anda menulis volume data besar ke tabel eksternal Parquet.

• Solusi

  Saat membuat tabel eksternal, pertama-tama kurangi parameter mcfed.parquet.block.row.count.limit. Jika error OOM masih terjadi atau file output terlalu besar, kurangi parameter mcfed.parquet.page.size.row.check.max untuk memeriksa memori lebih sering. Untuk informasi selengkapnya, lihat Parameter unik.

  Sebelum menulis data ke tabel eksternal Parquet, tambahkan parameter berikut.

  -- Atur ukuran memori maksimum untuk heap JVM UDF.
  SET odps.sql.udf.jvm.memory=12288;
  -- Kontrol ukuran batch di sisi waktu proses.
  SET odps.sql.executionengine.batch.rowcount =64;
  -- Atur ukuran memori untuk setiap pekerja Map.
  SET odps.stage.mapper.mem=12288;
  -- Atur volume data input untuk setiap pekerja Map (ukuran shard file input) untuk secara tidak langsung mengontrol jumlah pekerja per tahap Map.
  SET odps.stage.mapper.split.size=64;