Pyspark 를 사용한 테이블 조인, table join with pyspark

두 개의 HBase 테이블을 가지고 PySpark를 통해 통계 데이터를 생성하려면 다음과 같은 과정이 필요합니다. PySpark와 HBase를 연동하려면 Spark와 HBase 간의 통신을 설정하고, 테이블 데이터를 불러온 후 필요한 통계를 계산하는 로직을 구현해야 합니다.

아래는 PySpark와 HBase 연동 및 데이터 처리 과정에 대한 전체 흐름과 예제 코드입니다.

1. 환경 설정 및 의존성 추가

Spark와 HBase 간의 통신을 위해 몇 가지 의존성 패키지가 필요합니다:

  • hbase-client: HBase 클라이언트 라이브러리
  • hbase-spark: Spark와 HBase 간의 통합을 지원하는 모듈

다음과 같은 의존성을 Spark 환경에 추가해야 합니다:

spark-submit \
  --jars /path/to/hbase-spark.jar,/path/to/hbase-client.jar \
  your_spark_script.py

Spark 환경에 HBase 관련 의존성을 추가하지 않으면 PySpark 코드에서 HBase와의 연동이 제대로 동작하지 않습니다. 사용 중인 클러스터 환경에 따라 이 설정이 달라질 수 있습니다.

2. SparkContext 설정

PySpark에서 HBase와 연동하려면, hbase-site.xml 파일을 참조하거나 HBase의 설정을 명시적으로 지정해야 합니다. 이 예제에서는 SparkSession과 Hadoop 설정을 통해 HBase와의 연동을 준비합니다.

from pyspark.sql import SparkSession

# SparkSession 생성 (HBase와 통신 준비)
spark = SparkSession.builder \
    .appName("HBase-Stat-Job") \
    .config("spark.hadoop.hbase.zookeeper.quorum", "zookeeper-server") \
    .config("spark.hadoop.hbase.zookeeper.property.clientPort", "2181") \
    .config("spark.hadoop.hbase.mapreduce.inputtable", "table1") \
    .getOrCreate()

위 코드는 Spark 세션을 초기화하고, HBase의 ZooKeeper 설정을 추가한 것입니다. hbase-site.xml을 정확한 경로에 배치하면, Spark가 자동으로 HBase 클러스터와 연결될 수 있습니다.

3. HBase 테이블 데이터 로딩

HBase 테이블의 데이터를 PySpark DataFrame으로 불러오기 위해 다음과 같은 방식을 사용할 수 있습니다.

from pyspark.sql import DataFrame
from pyspark.sql.functions import col

def load_hbase_table(table_name: str) -> DataFrame:
    # HBase 테이블을 DataFrame으로 읽어오기
    return spark.read \
        .format("org.apache.hadoop.hbase.spark") \
        .option("hbase.table", table_name) \
        .option("hbase.spark.use.hbasecontext", "true") \
        .load()

# 두 개의 HBase 테이블을 불러오기
df1 = load_hbase_table("table1")
df2 = load_hbase_table("table2")

# 데이터 확인
df1.show()
df2.show()

위 코드에서는 HBase 테이블 데이터를 Spark DataFrame으로 변환합니다. table1과 table2의 데이터를 각각 df1과 df2에 로드한 예입니다.

4. 데이터 조인 및 통계 계산

두 테이블 간의 공통 키를 기준으로 조인한 후 필요한 통계 데이터를 계산할 수 있습니다.

예제: 두 테이블의 공통 키(id)를 기준으로 조인하고, 특정 컬럼에 대한 통계 계산

from pyspark.sql.functions import avg, count

# 'id'를 기준으로 두 테이블을 조인
joined_df = df1.join(df2, on="id", how="inner")

# 특정 컬럼에 대한 평균 및 개수 계산
stats_df = joined_df.groupBy("category").agg(
    avg("value1").alias("avg_value1"),
    avg("value2").alias("avg_value2"),
    count("id").alias("count")
)

# 통계 결과 출력
stats_df.show()

위 코드에서는 두 HBase 테이블을 id 컬럼을 기준으로 조인하고, category별로 그룹화한 후, value1과 value2 컬럼의 평균과 id의 개수를 계산합니다.

5. 결과 저장 (선택)

계산된 통계 데이터를 다시 HBase에 저장하거나, CSV/Parquet 파일로 저장할 수 있습니다.

HBase에 저장하는 예제

stats_df.write \
    .format("org.apache.hadoop.hbase.spark") \
    .option("hbase.table", "stats_table") \
    .save()

CSV 파일로 저장하는 예제

stats_df.write.csv("hdfs:///user/output/stats_data.csv")

6. 전체 코드 예시

from pyspark.sql import SparkSession
from pyspark.sql.functions import avg, count

# SparkSession 생성 및 HBase 설정
spark = SparkSession.builder \
    .appName("HBase-Stat-Job") \
    .config("spark.hadoop.hbase.zookeeper.quorum", "zookeeper-server") \
    .config("spark.hadoop.hbase.zookeeper.property.clientPort", "2181") \
    .getOrCreate()

# HBase 테이블을 DataFrame으로 불러오기 함수
def load_hbase_table(table_name: str):
    return spark.read \
        .format("org.apache.hadoop.hbase.spark") \
        .option("hbase.table", table_name) \
        .option("hbase.spark.use.hbasecontext", "true") \
        .load()

# 두 개의 테이블 불러오기
df1 = load_hbase_table("table1")
df2 = load_hbase_table("table2")

# 조인 및 통계 계산
joined_df = df1.join(df2, on="id", how="inner")
stats_df = joined_df.groupBy("category").agg(
    avg("value1").alias("avg_value1"),
    avg("value2").alias("avg_value2"),
    count("id").alias("count")
)

# 통계 결과 출력
stats_df.show()

# 통계 결과를 CSV로 저장
stats_df.write.csv("hdfs:///user/output/stats_data.csv")

# 세션 종료
spark.stop()

결론

위의 예제에서는 PySpark를 사용해 두 개의 HBase 테이블에서 데이터를 불러오고, 조인한 후 통계 데이터를 계산하는 과정을 보여드렸습니다. 사용자의 환경에 따라 HBase와의 연동 방식이 다를 수 있으니, Spark 및 HBase 클러스터 환경에 맞게 설정을 조정해야 합니다.

필요에 따라 결과 데이터를 다시 HBase에 저장하거나, 다른 저장소(CSV, Parquet 등)에 저장할 수 있습니다.

댓글

댓글 쓰기

이 블로그의 인기 게시물

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...
자세한 내용 보기

How to split a list into chunks of 100 items in JavaScript, 자바스크립트 리스트 쪼개기

To split a list into chunks of 100 items in JavaScript, you can create a function that loops through the list and slices it into smaller subarrays of size 100. Here’s how you can do it: Example Code: function splitListBy100( array ) { const chunkSize = 100 ; const result = []; for ( let i = 0 ; i < array .length; i += chunkSize) { result .push( array .slice(i, i + chunkSize)); } return result ; } // Example usage: const myArray = [... Array ( 350 ).keys()]; // Creating an array of numbers from 0 to 349 const chunks = splitListBy100(myArray); console.log(chunks); Explanation: chunkSize : Defines the size of each chunk, which in this case is 100. for loop : Iterates over the array, incrementing the index by 100 each time. array.slice(i, i + chunkSize) : Extracts a chunk from the array starting from index i and ending at i + chunkSize . result.push() : Adds the chunk to the result array. The function returns an array of arrays (...
자세한 내용 보기

HTML Inline divisions at one row by Tailwind

To arrange a **variable number of inline ` ` elements in a row** using **Tailwind CSS**, you can utilize several approaches, depending on your layout needs. Below are some common methods to align multiple divs in a row: ### 1. **Using Flexbox (`flex`)** Tailwind's `flex` utilities provide a straightforward way to align a variable number of ` ` elements inline in a row. Flexbox will ensure that they stay in a row, wrapping if necessary. #### **Example Using Flexbox** ```html Div 1 Div 2 Div 3 Div 4 ``` #### **Explanation:** - `flex`: Puts the child elements (the divs) in a row. - `flex-wrap`: Ensures that if there are too many divs to fit in one row, they will wrap to the next line. - `space-x-4`: Adds horizontal spacing between the divs (adjust as needed). ### 2. **Using Grid (`grid`)** Another way to align divs in a row is by using Tailwind's `grid` utilities. Grid layouts are very flexible and allow control over the number of columns. #### **Example Using ...
자세한 내용 보기

Boilerplate for typescript server programing

Here are some recommended boilerplates and frameworks for TypeScript server programming, depending on your requirements: 1. NestJS (Highly Recommended for Enterprise-Grade Applications) NestJS is a progressive Node.js framework built with TypeScript. It provides an opinionated structure and is excellent for building scalable and maintainable server-side applications. Features : Dependency injection Modular architecture Built-in support for WebSockets, GraphQL, REST APIs Easy integration with tools like Swagger, TypeORM, and Mongoose Getting Started : Install Nest CLI: npm i -g @nestjs/cli nest new project-name Boilerplate is automatically generated with ready-to-use configurations for controllers, services, and modules. Website : https://nestjs.com/ 2. Express with TypeScript Express.js is lightweight and minimal, making it ideal for developers who prefer flexibility over opinionated frameworks. Boilerplate : You can start with an official or community-maintai...
자세한 내용 보기

가속도 & 속도

F = ma 에서 가속도(( a ))는 속도(( v ))의 시간(( t ))에 대한 미분이므로, 수식으로 표현하면 다음과 같습니다. \[a = \frac{dv}{dt}\] 또한, 속도(( v ))는 위치(( x ))의 시간 미분이므로, 이를 포함하여 가속도를 표현하면 \[a = \frac{d}{dt} \left( \frac{dx}{dt} \right) = \frac{d^2 x}{dt^2}\] 즉, 가속도는 위치의 이阶 미분(두 번째 미분)입니다.
자세한 내용 보기

Gradle multi-module project

In a Gradle multi-module project , it's common to define shared dependencies in the parent project (root project) to avoid duplication across modules. This structure allows child modules to inherit or access those dependencies. Below are some best practices on how to correctly share dependencies between parent and child modules in a multi-module Gradle project . Project Structure Example multi-module-project/ ├── build .gradle .kts (Parent/root project) ├── settings .gradle .kts ├── module-a/ │ └── build .gradle .kts ├── module-b/ │ └── build .gradle .kts Step 1: Configure settings.gradle.kts In a multi-module project, the settings.gradle.kts file should declare all the modules that are part of the build. // settings.gradle.kts rootProject .name = "multi-module-project" // Include child modules include ( "module-a" , "module-b" ) Step 2: Define Shared Dependencies in the Parent build.gradle.kts The root build.gradle.kts ca...
자세한 내용 보기

How to checkout branch of remote git, 깃 리모트 브랜치 체크아웃

To check out a branch from a remote Git repository, you can follow these steps: 1. Fetch the remote branches First, ensure that your local repository is aware of the remote branches by using git fetch . This updates your local references to the remote branches. git fetch origin 2. Check out the remote branch Once the remote branches are fetched, you can check out the desired branch using git checkout -b and specifying the remote branch. Command: git checkout -b < local -branch- name > origin/<remote-branch- name > Explanation: <local-branch-name> : The name you want to give to your local branch. origin/<remote-branch-name> : Refers to the branch on the remote called origin (which is the default name for the main remote repository) and the branch you want to check out. Example: If the remote branch is called feature-branch and you want to create a local branch with the same name, you would run: git checkout -b feature -branch origin / feature -...
자세한 내용 보기

CDPEvents in puppeteer

In Puppeteer , CDPEvents refers to events emitted by the Chrome DevTools Protocol (CDP) . Puppeteer leverages CDP to interact with and control a Chromium-based browser. CDP provides detailed, low-level access to browser internals, such as network traffic, console logs, page lifecycle events, and more. You can listen to these CDP events through Puppeteer’s API to monitor or intercept browser activity. How to Listen for CDP Events in Puppeteer Enable the Required CDP Domain: Some events require enabling a particular domain (e.g., 'Network' , 'Page' , 'Runtime' ). Use page._client() to Access the CDP Session: Although it’s a bit lower-level, Puppeteer allows access to DevTools through the page._client() API. Example: Listening for Network Requests This example demonstrates how to intercept and log network requests using CDP. const puppeteer = require( 'puppeteer' ); ( async () => { const browser = await puppeteer.launch({ headless...
자세한 내용 보기

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...
자세한 내용 보기

Reactjs datetime range picker

To create a datetime range picker in React, you can use a library such as react-datetime-range-picker . This library provides a customizable date and time range picker component that you can easily integrate into your React application. Here's an example of how you can use it: Install the library using npm or yarn: npm install react-datetime- range -picker Import the necessary components into your React component: import React, { useState } from 'react' ; import DateTimeRangePicker from 'react-datetime-range-picker' ; Create a state to hold the selected date and time range: const [selectedRange, setSelectedRange] = useState({ startDate: null , endDate: null }); Render the DateTimeRangePicker component and handle the selection changes: ```jsx const handleRangeChange = (range) => { setSelectedRange(range); }; return ( ); ``` In the example above, the DateTimeRangePicker component is rendered with the onChange event handler and th...
자세한 내용 보기