PySpark Dataframe from HBase

To filter data from HBase and write it back using PySpark, follow these steps:

  1. Read data from HBase into a PySpark DataFrame.
  2. Filter the DataFrame using PySpark transformations.
  3. Write the filtered DataFrame to your desired format (e.g., HBase, Parquet, CSV, etc.).

Below is the complete example covering these tasks.

1. Reading, Filtering, and Writing Data

Step-by-Step Code

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# Step 1: Initialize Spark Session with HBase Configurations
spark = SparkSession.builder \
    .appName("HBase-PySpark Read-Write") \
    .config("spark.jars", "/path/to/hbase-client.jar,/path/to/hadoop-common.jar") \
    .config("spark.hadoop.hbase.zookeeper.quorum", "localhost") \
    .config("spark.hadoop.hbase.zookeeper.property.clientPort", "2181") \
    .getOrCreate()

# Step 2: Define the HBase Table Catalog
hbase_table = "my_table"
catalog = f"""
{{
    "table": {{
        "namespace": "default", 
        "name": "{hbase_table}"
    }},
    "rowkey": "key",
    "columns": {{
        "key": {{ "cf": "rowkey", "col": "key", "type": "string" }},
        "name": {{ "cf": "cf1", "col": "name", "type": "string" }},
        "age": {{ "cf": "cf1", "col": "age", "type": "int" }}
    }}
}}
"""

# Step 3: Read Data from HBase into a PySpark DataFrame
df = spark.read \
    .format("org.apache.hadoop.hbase.spark") \
    .option("catalog", catalog) \
    .load()

# Step 4: Filter the DataFrame (e.g., age > 20)
filtered_df = df.filter(col("age") > 20)

# Step 5: Show the Filtered Data
filtered_df.show()

# Step 6: Write Filtered Data to a New CSV File (or HBase, if needed)
filtered_df.write \
    .mode("overwrite") \
    .csv("output/filtered_hbase_data.csv")

2. Explanation of Steps

1. Spark Session Initialization

  • Configures Spark to interact with HBase using Zookeeper settings.
  • Loads the HBase and Hadoop client libraries (ensure the .jar paths are correct).

2. HBase Table Catalog

  • Defines how PySpark maps HBase columns to a DataFrame structure.

3. Filtering the Data

  • df.filter(col("age") > 20) filters out rows where the age is less than or equal to 20.

4. Writing Data

  • The write operation stores the filtered data in CSV format.
  • You can adjust the format to Parquet, JSON, or write it back to HBase.

3. Writing Data Back to HBase (Optional)

To write the filtered data back to HBase, update the catalog and use the HBase connector.

Example: Writing Data to HBase

# Define a new catalog for the target HBase table
write_catalog = f"""
{{
    "table": {{
        "namespace": "default", 
        "name": "filtered_table"
    }},
    "rowkey": "key",
    "columns": {{
        "key": {{ "cf": "rowkey", "col": "key", "type": "string" }},
        "name": {{ "cf": "cf1", "col": "name", "type": "string" }},
        "age": {{ "cf": "cf1", "col": "age", "type": "int" }}
    }}
}}
"""

# Write the filtered DataFrame back to HBase
filtered_df.write \
    .format("org.apache.hadoop.hbase.spark") \
    .option("catalog", write_catalog) \
    .mode("overwrite") \
    .save()

4. Summary

  • Read Data from HBase: Use PySpark with the HBase connector to load the data.
  • Filter Data: Apply filter() or where() clauses on the DataFrame.
  • Write Filtered Data: Save the data in CSV, Parquet, or write it back to HBase.

This approach ensures a smooth read, filter, and write pipeline using HBase with PySpark. Let me know if you run into any issues or need further clarification!

댓글

댓글 쓰기

이 블로그의 인기 게시물

Using the MinIO API via curl

Using the MinIO API via curl is straightforward, as MinIO is compatible with Amazon S3 API, so most commands follow a similar syntax. Here’s a guide on how to use curl with the MinIO API for some common operations like uploading, downloading, and managing objects. Prerequisites Access Key and Secret Key : Obtain your MinIO Access Key and Secret Key. MinIO Endpoint : Know your MinIO server endpoint, e.g., http://localhost:9000 . Bucket : You may need an existing bucket name, or create a new one using the commands below. Authentication Header For requests to work with MinIO, you need to include authentication in the headers. MinIO uses AWS Signature Version 4 for signing requests. Common Examples 1. List Buckets To list all buckets in your MinIO account, use: curl -X GET \ - -url "http://localhost:9000/" \ - H "Authorization: AWS <AccessKey>:<Signature>" 2. Create a Bucket To create a new bucket, use: curl -X PUT \ - -url "htt...
자세한 내용 보기

vsftpd default directory

The default directory for FTP users in vsftpd depends on how the FTP server is configured and the user’s home directory setup. Below are different configurations and their impact on the default directory that users are placed in when they connect to the server. 1. Default Directory Behavior with vsftpd When a user logs in via FTP, they are typically placed in their home directory . By default, the home directory is located at /home/<username> , where <username> is the name of the FTP user. For example: If the FTP user is ftpuser , the default login directory will be /home/ftpuser/ . 2. Changing the Default Directory for Users A. Using local_root Configuration You can specify a custom directory (other than the user’s home directory) for FTP users by setting the local_root parameter in the vsftpd configuration file. Open the vsftpd configuration file: sudo nano /etc/ vsftpd.conf Add the following line to specify a new directory for all FTP users: lo...
자세한 내용 보기

Using venv in Python

Using venv in Python allows you to create a virtual environment, which is an isolated environment to manage dependencies for a specific project. Here's how you can use it step by step: 1. Create a Virtual Environment Open your terminal or command prompt. Navigate to the folder where you want to create the virtual environment. Run the following command: python -m venv myenv Replace myenv with the name you want for your virtual environment. This will create a directory named myenv containing the virtual environment. 2. Activate the Virtual Environment Activation depends on your operating system: On Windows : myenv \S cripts \a ctivate On macOS/Linux : source myenv /bin/ activate After activation, your terminal prompt will change, showing the name of the virtual environment (e.g., (myenv) ). 3. Install Dependencies With the virtual environment activated, you can now install packages using pip : pip install package _n ame For example: pip install req...
자세한 내용 보기

Offset out of range error in Kafka, 카프카 트러블슈팅

The "Offset out of range" error in Kafka typically occurs when a consumer tries to read an offset that no longer exists on the broker. This can happen for several reasons, such as: The offset is smaller than the earliest offset available for the topic partition (i.e., the messages have been deleted due to log retention policies). The offset is larger than the latest offset available for the topic partition (i.e., the consumer is trying to read beyond the latest message). Steps to Resolve "Offset Out of Range" Error 1. Check Current Offsets Before making changes, it's useful to understand the current offsets and the offset range available for the topic partitions. Use the kafka-consumer-groups.sh script to describe the consumer group: bin/kafka-consumer-groups.sh --bootstrap- server localhost: 9092 --describe -- group <consumer- group -id> This command will show the current offsets, log end offsets, and the lag for each partition. 2. Reset the...
자세한 내용 보기

리눅스 (cron - 주기적 작업실행 데몬)

cron 데몬의 가동은 /etc/rc.d/init.d/crond 에 있다. 보통 시스템의 어떤 런레벨이든지 cron 데몬은 부팅 시 시작하도록 되어 있는데, 굳이 cron 데몬을 죽이거나 다시 시작하기 위해서는 위의 /etc/rc.d/init.d/crond 실행 스크립트를 실행하면 된다. cron 데몬에 작업을 설정하기 위해서는 cron table 을 설정하는 crontab 이라는 실행파일을 사용한다. cron table 을 편집하기 위해서는 crontab -e 명령을 사용한다. 입력 형식은 아래와 같다. #crontab -e 01 4 * * *  /etc/rc.d/init.d/network stop 15 4 * * * ~/scripts/my_script.sh 00 5 * * * /etc/rc.d/init.d/network start 위의 내용이 cron table 에 입력된 내용인데 이러한 입력형식은 아래와 같다. M H D m d cmd-line 필드명 범위 설명 M 0-59,* 분을 의미하며, *는 모든 볌위를 말함 H 0-23,* 시를 의미함 D 1-31,* 날짜를 의미함 m 1-12,* 달을 의미함 d 0-7,* 요일을 의미하며,0과 7은 일요일(Sun)을 의미함 cmd-line 실행할 명령을 입력 각자의 사용자가 등록한 cron table 은 /var/spool/cron 디렉토리에 저장된다. 입력 내용의 다른 예를 보면, 30  13  *  * 1-5  mail -s "It's 2pm" root% Dear root% % Are you sleeping?? 같은 경우는 월요일부터 금요일까지 오후 1시 30분에 해당 mail 명령을 실행하게 된다. 위에서 %문자는 개행문자 역할을 하며(풀어 쓰면 아래와 같은 명령이 된다.), 한줄이 넘어가는 명령을 쓰기 위해서는 백슬래쉬를 써주고 다시 개행해서 쓰면 된다. $mail -s "It's 2pm" ro...
자세한 내용 보기

[Ubuntu] *.deb 파일 설치 방법

확장자가 .deb 인 파일은 원래 Debian 리눅스에서의 설치파일입니다. Ubuntu 는 원래 Debian 리눅스에서 파생되었기 때문에, 많은 부분에서 서로 닮아 있습니다. 따라서 별다른 변환 절차 없이 아래와 같이 바로 설치가 가능합니다. [user@hostname]# sudo dpkg -i *.deb *.rpm 파일을 *.deb 파일로 변환 [user@hostname]$ sudo apt-get install alien [user@hostname]$ sudo alien -c *.rpm
자세한 내용 보기

Sparse encoder

A Sparse Encoder refers to a variant of neural network architectures where sparsity is introduced in the encoding process. This can mean either: Sparse Input Representations : The input features to the encoder are sparse (many values are zero). Sparse Output Representations : The encoder is designed to produce sparse outputs where most of the encoded feature values are zero. Sparse encoders are often used to improve model interpretability, efficiency, and generalization. They can be applied in various contexts, including traditional neural networks, autoencoders, and even transformer-based models. Key Characteristics of Sparse Encoders Sparsity in Representations : The model learns a feature representation where only a subset of neurons is active for a given input. This mimics how biological neurons operate, promoting interpretability and reducing noise in representations. Reduced Computational Cost : Sparse operations often result in lower computational overhead since...
자세한 내용 보기

Retrieving Open Graph (OG) tags from an HTML document in Java

Retrieving Open Graph (OG) tags from an HTML document in Java typically involves fetching the HTML content of a web page and then parsing it to extract the <meta> tags with the property attribute starting with og: . Here's how you can do it step by step: 1. Add Required Dependencies You can use libraries like Jsoup for HTML parsing. Add the following dependency to your pom.xml if you're using Maven: < dependency > < groupId > org.jsoup </ groupId > < artifactId > jsoup </ artifactId > < version > 1.16.1 </ version > <!-- Use the latest version --> </ dependency > 2. Code to Retrieve OG Tags Here's a sample Java program to fetch and parse OG tags: import org.jsoup.Jsoup; import org.jsoup.nodes.Document; import org.jsoup.nodes. Element ; import org.jsoup.select.Elements; import java.io.IOException; import java.util.HashMap; import java.util. Map ; public class OGTagExtractor { ...
자세한 내용 보기