在Apache DataHub中整合Amazon Glue任务的数据血缘

sudo yum install docker
curl -SL https://github.com/docker/compose/releases/download/v2.39.3/docker-compose-linux-x86_64 -o  docker-compose
sudo mv docker-compose /usr/local/bin/
chmod +x /usr/local/bin/docker-compose
mkdir -p ~/miniconda3
wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh -O ~/miniconda3/miniconda.sh
bash ~/miniconda3/miniconda.sh -b -u -p ~/miniconda3
rm ~/miniconda3/miniconda.sh
source ~/miniconda3/bin/activate
conda init --all
conda create --name datahub python=3.11
conda activate datahub

python3 -m pip install --upgrade pip wheel setuptools
python3 -m pip install --upgrade acryl-datahub
datahub version

from datahub.ingestion.run.pipeline import Pipeline

# The pipeline configuration is similar to the recipe YAML files provided to the CLI tool.
pipeline = Pipeline.create(
    {
        "source": {
            "type": "glue",
            "config": {
                "aws_region": "cn-north-1",
                "emit_s3_lineage" : False,
                "extract_transforms": False

            },
        },
        "sink": {
            "type": "datahub-rest",
            "config": {
                "server": "http://ip-172-31-xxx-xxx:8080",
                 "token": "xxxxxxxxxxxxxxxxxxDg"
                },
        },
    }
)

# Run the pipeline and report the results.
pipeline.run()
pipeline.pretty_print_summary()

CREATE EXTERNAL TABLE `tripdata`(
  `vendorid` bigint, 
  `tpep_pickup_datetime` timestamp, 
  `tpep_dropoff_datetime` timestamp, 
  `passenger_count` double, 
  `trip_distance` double, 
  `ratecodeid` double, 
  `store_and_fwd_flag` string, 
  `pulocationid` bigint, 
  `dolocationid` bigint, 
  `payment_type` bigint, 
  `fare_amount` double, 
  `extra` double, 
  `mta_tax` double, 
  `tip_amount` double, 
  `tolls_amount` double, 
  `improvement_surcharge` double, 
  `total_amount` double, 
  `congestion_surcharge` double, 
  `airport_fee` double)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://{your_bucket}/tripdata/'

CREATE EXTERNAL TABLE `curated_nyx_taxi`(
  `vendorid` bigint, 
  `pickup_datetime` timestamp, 
  `dropoff_datetime` timestamp, 
  `passenger_count` double, 
  `trip_distance` double, 
  `ratecodeid` double, 
  `store_and_fwd_flag` string, 
  `pulocationid` bigint, 
  `dolocationid` bigint, 
  `payment_type` bigint, 
  `fare_amount` double, 
  `extra` double, 
  `mta_tax` double, 
  `tip_amount` double, 
  `tolls_amount` double, 
  `improvement_surcharge` double, 
  `total_amount` double, 
  `congestion_surcharge` double, 
  `airport_fee` double)
PARTITIONED BY ( 
  `pickup_date` date)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://{your_bucket}/tripdata_output/'

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job

from pyspark.sql.functions import udf, col
from pyspark.sql.types import IntegerType, StringType
from pyspark.sql import SQLContext
from pyspark.context import SparkContext
from datetime import datetime
from pyspark.sql.session import SparkSession


from pyspark.sql import  Row
import pyspark.sql.functions as F
import pyspark.sql.types as T
import pyspark.sql.window as W

spark = SparkSession.builder.config("spark.extraListeners", "datahub.spark.DatahubSparkListener") \
.config("spark.datahub.rest.server","<GMS_ENDPOINT>") \
.config("spark.datahub.rest.token","<GMS_TOKEN>") \
.config("spark.app.name","datahubbloggluelineage") \
.getOrCreate()

spark.conf.set('spark.sql.sources.partitionOverwriteMode', 'dynamic')
spark.conf.set('hive.exec.dynamic.partition', 'true')
spark.conf.set('hive.exec.dynamic.partition.mode', 'nonstrict')

glueContext = GlueContext(spark.sparkContext)
job = Job(glueContext)

logger = glueContext.get_logger()
logger.info("Job initializied")

args = getResolvedOptions(sys.argv, ['JOB_NAME'])
job.init(args['JOB_NAME'], args)
job_time_string = datetime.now().strftime("%Y%m%d%H%M%S")

landing_table_name = "datahub_nyx_taxi.landing_nyx_taxi"

spark_new_df = spark.sql("select * from "+landing_table_name)
spark_new_df.printSchema()

spark_df_transform = spark_new_df.withColumn("pickup_date",F.to_date("tpep_pickup_datetime"))\
                                 .withColumn("pickup_datetime",F.to_timestamp("tpep_pickup_datetime"))\
                                 .withColumn("dropoff_datetime",F.to_timestamp("tpep_dropoff_datetime"))

spark_df_transform = spark_df_transform.drop("tpep_pickup_datetime","tpep_dropoff_datetime")

target_table_name = 'datahub_nyx_taxi.curated_nyx_taxi'
cols = spark.sql("select * from " + target_table_name+ " where 1=0").columns
spark_df_transform = spark_df_transform.select(cols)
  

spark_df_transform.write.mode("overwrite").insertInto(target_table_name)
job.commit()