PySpark

Giới Thiệu Apache Hive là một hệ thống lưu trữ dữ liệu được xây dựng trên nền tảng của Hadoop, hỗ trợ phân tích các tập dữ liệu lớn được lưu trữ trong các hệ thống tệp phân tán của Hadoop (HDFS) hoặc S3 bằng cách sử dụng các truy vấn tương tự SQL. Nó lưu trữ siêu dữ liệu (metadata) của schema trong một cơ sở dữ liệu (như Apache Derby, MySQL, Postgres hoặc MariaDB) và xử lý dữ liệu từ lưu trữ kết nối (như HDFS hoặc S3). Nó cung cấp ngôn ngữ truy vấn giống SQL, HiveQL. Hive Metastore Hive metastore ghi lại tất cả thông tin cấu trúc (metadata) của một bảng Hive và các phân vùng của nó trong kho lưu trữ: Các siêu dữ liệu trong Hive metastore được lưu trữ trong một hệ quản trị cơ sở dữ liệu mã nguồn mở là Apache Derby (MySQL/Postgres/MariaDB cũng có thể được sử dụng). Hive metastore có thể có nhiều schema (cơ sở dữ liệu), và mỗi schema (cơ sở dữ liệu) có thể chứa nhiều siêu dữ liệu bảng Hive. Bảng Hive Loại bảng Hive Hive hỗ trợ hai loại bảng: Bảng Quản lý (Bảng Nội bộ) & Bảng Ngoại bộ Bảng Quản lý (Bảng Nội bộ) Bảng Quản lý: còn được gọi là bảng nội bộ, các bảng này quản lý cả dữ liệu và siêu dữ liệu trong thư mục kho lưu trữ mặc định của Hive (được chỉ định bởi hive.metastore.warehouse.dir). Bảng Ngoại bộ Bảng Ngoại bộ: các bảng này tham chiếu đến các tệp dữ liệu được lưu trữ bên ngoài Hive, thường là trong HDFS hoặc S3. Vì các tệp dữ liệu không được quản lý bởi Hive, nên việc thay đổi hoặc xóa một bảng Hive ngoại bộ không xóa dữ liệu ở dạng dưới lying. Bạn có thể cần xóa dữ liệu từ vị trí HDSF hoặc S3 phía dưới. Cách Tạo Bảng Hive Bước 1: Tạo SparkSession với cấu hình Hive pythonSao chép mãspark = SparkSession.builder \ .appName(“Hive Table Example”) \ .config(“spark.sql.warehouse.dir”, “/user/hive/warehouse”) \ .enableHiveSupport() \ .getOrCreate() Bước 2: Có hai phương pháp để tạo bảng Hive Phương pháp 1: Tạo bảng Hive bằng cách sử dụng API DataFrame với write.saveAsTable pythonSao chép mã# Bảng Quản lý (Bảng Nội bộ) df.write \ .mode(“overwrite”) \ .option(“database”, “your_database_name”) \ .partitionBy(“partition_col1”, “partition_col2”) \ .saveAsTable(“hive_table_name”) # Bảng Ngoại bộ df.write \ .mode(“overwrite”) \ .option(“database”, “your_database_name”) \ .option(“path”, “s3a://your_bucket_name/your_folder_path/your_external_table_name”) \ .partitionBy(“partition_col1”, “partition_col2”) \ .format(“parquet”) \ .saveAsTable(“hive_table_name”) Phương pháp 2: Tạo Bảng Hive bằng cách sử dụng Truy vấn SQL với spark.sql pythonSao chép mã# Bảng Quản lý (Bảng Nội bộ) spark.sql(“”” CREATE TABLE IF NOT EXISTS your_database_name.your_managed_table_name ( id INT, name STRING, age INT ) PARTITIONED BY (country STRING, city STRING) USING PARQUET “””) # Bảng Ngoại bộ spark.sql(“”” CREATE TABLE IF NOT EXISTS your_database_name.your_external_table_name ( id INT, name STRING, age INT ) PARTITIONED BY (country STRING, city STRING) STORED AS PARQUET LOCATION ‘hdfs://your_hdfs_path/your_external_table_name/’ “””) Cách chèn dữ liệu vào Bảng Hive Phương pháp 1: Ghi trực tiếp thông qua write.saveAsTable vào Bảng Hive pythonSao chép mãdf.write \ .mode(“overwrite”) \ .partitionBy(“partition_col1”, “partition_col2”) \ .format(“parquet”) \ .saveAsTable(“your_database_name.your_hive_table_name”) Phương pháp 2: Ghi vào vị trí lưu trữ dưới lying & cập nhật lại siêu dữ liệu Hive pythonSao chép mãtable_path = “s3a://your_bucket_name/your_folder_path/your_hive_table_name” df.write \ .mode(“overwrite”) \ .partitionBy(“partition_col1”, “partition_col2”) \ .parquet(table_path) spark.sql(“REFRESH TABLE your_database_name.your_hive_table_name”) spark.sql(“MSCK REPAIR your_database_name.your_hive_table_name”) Cách Xem Chi Tiết Bảng Hive Lấy thông tin chi tiết về bảng pythonSao chép mãtable_metadata = spark.sql(“DESCRIBE FORMATTED db.table_name”) table_metadata.show(truncate=False) Lấy thông tin về phân vùng của bảng pythonSao chép mãpartitions = spark.sql(“SHOW PARTITIONS db.table_name”) partitions.show(truncate=False) Đây là một cách sơ bộ để làm việc với Hive trong Spark. Có nhiều khía cạnh và tính năng hơn mà bạn có thể khám phá!

Tạo Temp View Bạn có thể đăng ký bất kỳ DataFrame nào như một bảng hoặc view (một bảng tạm thời) và truy vấn nó bằng cách sử dụng SQL thuần túy với hàm spark.sql (nhớ, spark là biến SparkSession của chúng ta). pythonSao chép mãfrom pyspark.sql import SparkSession # Tạo một SparkSession spark = SparkSession.builder.appName(“TempViewExample”).getOrCreate() # Đăng ký DataFrame là một Temp View df.createOrReplaceTempView(“flight_data”) # Truy vấn Temp View bằng SQL sql_way = spark.sql(“”” SELECT DEST_COUNTRY_NAME, count(1) FROM flight_data GROUP BY DEST_COUNTRY_NAME “””) # Truy vấn DataFrame bằng cách sử dụng API DataFrame df_way = df \ .groupBy(“DEST_COUNTRY_NAME”)\ .count() Kiểm tra số lượng Temp Views pythonSao chép mã# Lấy danh sách các Temp Views từ catalog temp_views_list = spark.catalog.listTables() # Đếm số lượng Temp Views num_temp_views = len(temp_views_list) # In ra danh sách Temp Views và số lượng print(f”Danh sách {num_temp_views} Temp Views:”) for temp_view in temp_views_list: print(temp_view.name) createOrReplaceTempView và registerTempTable Cả createOrReplaceTempView và registerTempTable đều được sử dụng để tạo các view tạm thời từ DataFrame, làm cho chúng có sẵn cho các truy vấn SQL. Tuy nhiên, createOrReplaceTempView là phương pháp được khuyến nghị để tạo các view tạm thời trong Apache Spark, cung cấp tính tương thích và chức năng tốt hơn so với registerTempTable đã lỗi thời.

Giới thiệu SparkSession là gì? Bước đầu tiên trong bất kỳ Ứng dụng Spark nào là tạo một SparkSession. SparkSession hoạt động như quá trình điều khiển để kiểm soát ứng dụng Spark. Khi bạn khởi đầu Spark trong chế độ tương tác, một SparkSession được tạo một cách ngầm định để quản lý Ứng dụng Spark. Tuy nhiên, khi khởi động nó thông qua một ứng dụng độc lập, bạn cần tự tạo đối tượng SparkSession trong mã ứng dụng của bạn. SparkSession so với SparkContext Trong Spark 1.X, có hai ngữ cảnh hiệu quả: SparkContext và SQLContext. SparkContext tập trung vào kiểm soát tinh tế hơn của các trừng tượng trung tâm của Spark, trong khi SQLContext xử lý các công cụ cấp cao hơn như Spark SQL. Tuy nhiên, trong Spark 2.X, cộng đồng đã hợp nhất hai API này thành SparkSession tập trung để quản lý tốt hơn và ngăn chặn xung đột ngữ cảnh. Mặc dù SparkContext và SQLContext vẫn có sẵn trong SparkSession, nhưng hiếm khi được sử dụng trực tiếp. Tạo SparkSession Một SparkSession là điểm vào tất cả các chức năng trong Spark và là bắt buộc nếu bạn muốn xây dựng một DataFrame trong PySpark. pythonSao chép mãspark = SparkSession \ .builder \ .appName(“SparkExample”) \ .enableHiveSupport() \ # Để làm việc với Hive metastore. .config(“spark.memory.offHeap.enabled”,”true”) \ # Bộ nhớ cache dữ liệu trong bộ nhớ off-heap để tránh lưu trữ nó trực tiếp trên đĩa. .config(“spark.memory.offHeap.size”,”10g”) \ .getOrCreate() # Tạo Spark session với các cấu hình đã chỉ định. Một ví dụ khác với nhiều cấu hình hơn: pythonSao chép mãdef create_spark_session(): ”’ Hàm để tạo một Spark session. Output: SparkSession ”’ try: spark = SparkSession \ .builder \ .appName(“SparkExample”) \ .enableHiveSupport() \ # Để làm việc với Hive metastore. .config(“spark.driver.memory”, “9g”) \ .config(“spark.executor.cores”, “2”) \ .config(“spark.executor.memory”, “9g”) \ .config(“spark.yarn.queue”, “root.tnm.ada_analytics_tnm”) \ # Cần cập nhật với tên Queue YARN. custom_spark_config = { “spark.driver.memory”: “9g”, “spark.executor.cores”: “2”, “spark.executor.memory”: “9g”, “spark.yarn.queue”: “root.tnm.ada_analytics_tnm”, # Cần cập nhật với tên Queue YARN. } for key, value in custom_spark_config.items(): spark = spark.config(key, value) spark = spark.getOrCreate() # Tạo Spark session với các cấu hình đã chỉ định. return spark except RuntimeError as e: print(‘Lỗi khi tạo Spark context’) print(e) Hàm này tạo một Spark session với các cấu hình đã chỉ định, hỗ trợ kiểm soát tinh tế hơn đối với ứng dụng Spark.

Apache Spark là gì? Apache Spark là một công cụ tính toán thống nhất và một tập hợp các thư viện được thiết kế để xử lý dữ liệu song song trên các cụm máy tính. Nó có thể quản lý các cụm máy tính bằng một trình quản lý cụm như trình quản lý cụm độc lập của Spark, YARN hoặc Mesos. Các ứng dụng Spark được nộp lên các trình quản lý cụm này, sau đó sẽ phân bổ các tài nguyên cần thiết để thực hiện các tác vụ. Spark chủ yếu được viết bằng ngôn ngữ Scala và chạy trên Java Virtual Machine (JVM). Do đó, để chạy Spark trên máy tính cá nhân hoặc một cụm máy tính, bạn cần cài đặt Java. Nếu bạn muốn sử dụng API Python, cũng cần có một trình thông dịch Python. Các thuật ngữ quan trọng trong PySpark Các thành phần của Ứng dụng Spark Kiến trúc của Spark Trong một Ứng dụng Spark, trình quản lý cụm kiểm soát các máy vật lý và phân bổ tài nguyên cho các ứng dụng. Bắt đầu một Ứng dụng Spark Các chế độ chạy của Ứng dụng Spark

1. Giới thiệu Hive Hive Apache Hive là hệ thống lưu trữ dữ liệu được xây dựng trên nền tảng của Hadoop, hỗ trợ phân tích các tập dữ liệu lớn được lưu trữ trong hệ thống tệp phân tán Hadoop (HDFS) hoặc S3 bằng cách sử dụng các truy vấn giống như SQL. Nó lưu trữ siêu dữ liệu của schema trong cơ sở dữ liệu (như Postgres) và xử lý dữ liệu từ lưu trữ kết nối (như HDFS hoặc S3). Nó cung cấp ngôn ngữ truy vấn giống như SQL, HiveQL. Hệ thống lưu trữ Hive Hệ thống lưu trữ Hive ghi lại tất cả thông tin cấu trúc (metadata) của một bảng hive và các phân vùng của nó trong kho dữ liệu: – Thông tin cột và định nghĩa kiểu dữ liệu – Các bộ chuyển đổi cần thiết để đọc và ghi dữ liệu – Vị trí của dữ liệu liên quan trong HDFS hoặc S3 Metadata trong hệ thống lưu trữ Hive được lưu trữ trong một RDBMS mã nguồn mở là Apache Derby (cũng có thể sử dụng MySQL/ Postgres/ MariaDB) Hệ thống lưu trữ Hive có thể có nhiều cơ sở dữ liệu, và mỗi cơ sở dữ liệu có thể chứa nhiều bảng Hive. Dưới đây là mã pyspark để liệt kê tất cả cơ sở dữ liệu có sẵn trong hệ thống lưu trữ Hive và các bảng Hive trong mỗi cơ sở dữ liệu # Tùy chọn 1: Sử dụng phiên Spark để liệt kê cơ sở dữ liệu và bảng Hive có sẵn trong hệ thống lưu trữ Hive# Liệt kê tất cả cơ sở dữ liệudatabases = spark.catalog.listDatabases()print(“List of Hive Databases:”)for db in databases: print(db.name)# Liệt kê tất cả các bảng trong mỗi cơ sở dữ liệufor db in databases: print(f”\nTables in database: {db.name}”) tables = spark.catalog.listTables(db.name) for table in tables: print(f”\t{table.name}”) # Lựa chọn 2: sử dụng spark.sql để liệt kê cơ sở dữ liệu & bảng hive có sẵn trong Hive metastore# Liệt kê tất cả các cơ sở dữ liệudatabases_df = spark.sql(“SHOW DATABASES”)databases = databases_df.select(“databaseName”).rdd.flatMap(lambda x: x).collect() # flatten the databases to a list# Duyệt qua từng cơ sở dữ liệu và liệt kê các bảng của nófor database in databases: spark.sql(f”USE {database}”) tables_df = spark.sql(“SHOW TABLES”) tables = tables_df.select(“tableName”).rdd.flatMap(lambda x: x).collect() print(f”Database: {database}”) print(“Tables:”) for table in tables: print(f” {table}”) Bảng Hive Hive hỗ trợ hai loại bảng: Bảng Quản lý (Bảng Nội bộ) & Bảng Ngoại vi Bảng Quản lý – Managed Tables (Bảng Nội bộ) Còn được gọi là bảng nội bộ, các bảng này quản lý cả dữ liệu và siêu dữ liệu trong thư mục kho mặc định của Hive (được chỉ định bởi hive.metastore.warehouse.dir). Bảng Ngoại vi Các bảng này tham chiếu đến các tệp dữ liệu được lưu trữ bên ngoài Hive, thường là trong HDFS hoặc S3. – Vì các tệp dữ liệu không được Hive quản lý, nên việc thay đổi hoặc xóa một bảng ngoại vi của Hive không xóa dữ liệu cơ bản. Bạn có thể phải xóa dữ liệu qua vị trí HDFS hoặc S3 cơ bản. Bước 1: tạo SparkSession với cấu hình Hive spark.sql.warehouse.dir là thuộc tính cấu hình chỉ định vị trí mặc định nơi Spark SQL sẽ lưu trữ các bảng quản lý trong thư mục kho của Hive. Các bảng quản lý là những bảng mà dữ liệu và siêu dữ liệu được Spark quản lý, và các bảng này được tạo bằng lệnh CREATE TABLE SQL Nếu bạn không thiết lập cấu hình này rõ ràng, Spark sẽ sử dụng vị trí kho mặc định của Hive, thường được đặt là /user/hive/warehouse