AWS Data Engineering Bootcamp شرح ببساطة (مع مخططات وكود حقيقي)

-- الخطوة 1: إعداد التجميع أثناء إنشاء الجدول
-- استبدل user_id بعمودك ذي الكاردينالية العالية
CLUSTERED BY (user_id) INTO 8 BUCKETS

-- مثال كامل باستخدام CTAS (Create Table As Select)
CREATE TABLE taxidata.transactions_bucketized
WITH (
    format = 'PARQUET',
    external_location = 's3://mon-bucket/transactions-bucketized/',
    bucketed_by = ARRAY['user_id'],
    bucket_count = 8
) AS
SELECT *
FROM taxidata.transactions_source;

-- CTAS (Create Table As Select) مع التقسيم
-- تحويل CSV → Parquet مع التقسيم
CREATE TABLE taxidata.yellow_partitioned
WITH (
    format = 'PARQUET',
    external_location = 's3://taxi-datalake/yellow-partitioned/',
    partitioned_by = ARRAY['year', 'month']
) AS
SELECT
    vendor, pickup, dropoff, count,
    distance, ratecode, storeflag, pulocid, dolocid,
    paytype, fare, extra, mta_tax, tip, tolls, surcharge, total,
    YEAR(pickup) AS year,
    MONTH(pickup) AS month
FROM taxidata.yellow;

-- بعد إضافة البيانات، اكتشف الأقسام الجديدة
MSCK REPAIR TABLE taxidata.yellow_partitioned;

-- إضافة قسم محدد يدوياً
ALTER TABLE taxidata.yellow_partitioned
ADD PARTITION (year='2017', month='02')
LOCATION 's3://taxi-datalake/yellow-partitioned/year=2017/month=02/';

-- استعلام بدون تقسيم: مسح كامل الجدول
SELECT COUNT(*), SUM(fare) AS total_fares
FROM taxidata.yellow
WHERE YEAR(pickup) = 2017 AND MONTH(pickup) = 1;
-- → Scan : 100% des données (ex: 1 TB)

-- استعلام مع التقسيم: مسح القسم المعني فقط
SELECT COUNT(*), SUM(fare) AS total_fares
FROM taxidata.yellow_partitioned
WHERE year = 2017 AND month = 1;
-- → Scan : ~1/36 des données (si 3 ans × 12 mois)
-- → Réduction de ~97% des données scannées !

-- إنشاء قاعدة بيانات taxidata في AWS Glue Data Catalog
CREATE DATABASE taxidata;

-- إنشاء الجدول الخارجي yellow على بيانات S3
CREATE EXTERNAL TABLE IF NOT EXISTS taxidata.yellow (
   `vendor`    string,
   `pickup`    timestamp,
   `dropoff`   timestamp,
   `count`     int,
   `distance`  int,
   `ratecode`  string,
   `storeflag` string,
   `pulocid`   string,
   `dolocid`   string,
   `paytype`   string,
   `fare`      decimal,
   `extra`     decimal,
   `mta_tax`   decimal,
   `tip`       decimal,
   `tolls`     decimal,
   `surcharge` decimal,
   `total`     decimal
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
WITH SERDEPROPERTIES (
   'serialization.format' = ',',
   'field.delim' = ','
) LOCATION 's3://aws-tc-largeobjects/CUR-TF-200-ACDSCI-1/Lab2/yellow/'
TBLPROPERTIES ('has_encrypted_data'='false');

-- معاينة أول 10 صفوف
SELECT * FROM taxidata.yellow LIMIT 10;

-- عد إجمالي عدد الرحلات
SELECT COUNT(*) AS total_trajets FROM taxidata.yellow;

-- التحقق من أنواع الدفع المتاحة
SELECT DISTINCT paytype, COUNT(*) AS nb_trajets
FROM taxidata.yellow
GROUP BY paytype
ORDER BY nb_trajets DESC;

-- إنشاء جدول لبيانات يناير 2017 فقط
-- أسرع للاختبار والتحقق
CREATE EXTERNAL TABLE IF NOT EXISTS taxidata.jan (
   `vendor`    string,
   `pickup`    timestamp,
   `dropoff`   timestamp,
   `count`     int,
   `distance`  int,
   `ratecode`  string,
   `storeflag` string,
   `pulocid`   string,
   `dolocid`   string,
   `paytype`   string,
   `fare`      decimal,
   `extra`     decimal,
   `mta_tax`   decimal,
   `tip`       decimal,
   `tolls`     decimal,
   `surcharge` decimal,
   `total`     decimal
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
WITH SERDEPROPERTIES (
   'serialization.format' = ',',
   'field.delim' = ','
) LOCATION 's3://aws-tc-largeobjects/CUR-TF-200-ACDSCI-1/Lab2/January2017/'
TBLPROPERTIES ('has_encrypted_data'='false');

-- إنشاء طريقة عرض للمدفوعات ببطاقة الائتمان
-- paytype = '1' تعني الدفع ببطاقة
CREATE VIEW taxidata.cctrips AS
SELECT
    vendor,
    SUM(fare)  AS total_fares_cc,
    COUNT(*)   AS nb_trajets_cc,
    AVG(fare)  AS tarif_moyen_cc
FROM taxidata.yellow
WHERE paytype = '1'
GROUP BY vendor;

-- استخدام طريقة العرض
SELECT * FROM taxidata.cctrips;

-- إنشاء طريقة عرض للمدفوعات النقدية
-- paytype = '2' تعني الدفع نقداً
CREATE VIEW taxidata.cashtrips AS
SELECT
    vendor,
    SUM(fare)  AS total_fares_cash,
    COUNT(*)   AS nb_trajets_cash,
    AVG(fare)  AS tarif_moyen_cash
FROM taxidata.yellow
WHERE paytype = '2'
GROUP BY vendor;

-- استخدام طريقة العرض
SELECT * FROM taxidata.cashtrips;

-- طريقة عرض مقارنة تجمع بين بطاقة الائتمان والنقد
-- استخدام CTE (Common Table Expressions) مع WITH
CREATE VIEW taxidata.comparepay AS
WITH
    cc AS (
        SELECT
            vendor,
            SUM(fare) AS cctotal,
            COUNT(*)  AS nb_cc
        FROM taxidata.yellow
        WHERE paytype = '1'
        GROUP BY vendor
    ),
    cash AS (
        SELECT
            vendor,
            SUM(fare) AS cashtotal,
            COUNT(*)  AS nb_cash
        FROM taxidata.yellow
        WHERE paytype = '2'
        GROUP BY vendor
    )
SELECT
    cc.vendor,
    cc.cctotal,
    cc.nb_cc,
    cash.cashtotal,
    cash.nb_cash,
    ROUND(cc.cctotal / cash.cashtotal, 2) AS ratio_cc_cash
FROM cc
JOIN cash ON cc.vendor = cash.vendor;

-- استعلام على طريقة العرض
SELECT * FROM taxidata.comparepay ORDER BY ratio_cc_cash DESC;

-- سرد طرق العرض المتاحة
SHOW VIEWS IN taxidata;

-- عرض تعريف طريقة عرض
SHOW CREATE VIEW taxidata.comparepay;

-- استبدال طريقة عرض موجودة
CREATE OR REPLACE VIEW taxidata.cctrips AS
SELECT
    vendor,
    SUM(fare) AS total_fares_cc,
    COUNT(*) AS nb_trajets_cc
FROM taxidata.yellow
WHERE paytype = '1'
GROUP BY vendor;

-- حذف طريقة عرض
DROP VIEW taxidata.cashtrips;

AWSTemplateFormatVersion: 2010-09-09
Description: "Requête Athena nommée pour l'analyse des taxis NYC"

Resources:
  AthenaNamedQuery:
    Type: AWS::Athena::NamedQuery
    Properties:
      Database: "taxidata"
      Description: "Sélection des trajets dont le montant dépasse 100 $"
      Name: "FaresOver100DollarsUS"
      QueryString: >
        SELECT
            distance,
            paytype,
            fare,
            tip,
            tolls,
            surcharge,
            total
        FROM yellow
        WHERE total >= 100.0
        ORDER BY total DESC

Outputs:
  AthenaNamedQuery:
    Description: "L'ID de la requête nommée Athena créée"
    Value: !Ref AthenaNamedQuery