Databricks의 프로세스 자동화를 통한 비용 절감

저는 클라이언트에서 Databricks에서 실행되는 프로세스 비용을 절감해야 했습니다. Databricks가 담당한 기능 중 하나는 다양한 SFTP에서 파일을 수집하고 압축을 풀어 데이터 레이크에 배치하는 것이었습니다.

데이터 워크플로 자동화는 현대 데이터 엔지니어링에서 중요한 구성 요소입니다. 이 기사에서는 Go 애플리케이션이 SFTP 서버에 연결하고, 파일을 수집하고, Amazon S3에 저장하고, 마지막으로 Databricks에서 작업을 트리거할 수 있도록 GitLab CI/CD 및 Terraform을 사용하여 AWS Lambda 함수를 생성하는 방법을 살펴보겠습니다. 이러한 엔드투엔드 프로세스는 효율적인 데이터 통합 ​​및 자동화에 의존하는 시스템에 필수적입니다.

이 기사에 필요한 것

• 프로젝트 저장소가 있는 GitLab 계정입니다.
• Lambda, S3 및 IAM 리소스를 생성할 권한이 있는 AWS 계정.
• 작업을 생성하고 실행할 수 있는 권한이 있는 Databricks 계정입니다.
• Go, Terraform, GitLab CI/CD에 대한 기본 지식

1단계: Go 애플리케이션 준비

SFTP 서버에 연결하여 파일을 수집하는 Go 애플리케이션을 만드는 것부터 시작하세요. github.com/pkg/sftp와 같은 패키지를 사용하여 SFTP 연결을 설정하고 github.com/aws/aws-sdk-go를 사용하여 AWS S3 서비스와 상호 작용합니다.

package main

import (
 "fmt"
 "log"
 "os"
 "path/filepath"

 "github.com/pkg/sftp"
 "golang.org/x/crypto/ssh"
 "github.com/aws/aws-sdk-go/aws"
 "github.com/aws/aws-sdk-go/aws/session"
 "github.com/aws/aws-sdk-go/service/s3/s3manager"
)

func main() {
 // Configuração do cliente SFTP
 user := "seu_usuario_sftp"
 pass := "sua_senha_sftp"
 host := "endereco_sftp:22"
 config := &ssh.ClientConfig{
  User: user,
  Auth: []ssh.AuthMethod{
   ssh.Password(pass),
  },
  HostKeyCallback: ssh.InsecureIgnoreHostKey(),
 }

 // Conectar ao servidor SFTP
 conn, err := ssh.Dial("tcp", host, config)
 if err != nil {
  log.Fatal(err)
 }
 client, err := sftp.NewClient(conn)
 if err != nil {
  log.Fatal(err)
 }
 defer client.Close()

 // Baixar arquivos do SFTP
 remoteFilePath := "/path/to/remote/file"
 localDir := "/path/to/local/dir"
 localFilePath := filepath.Join(localDir, filepath.Base(remoteFilePath))
 dstFile, err := os.Create(localFilePath)
 if err != nil {
  log.Fatal(err)
 }
 defer dstFile.Close()

 srcFile, err := client.Open(remoteFilePath)
 if err != nil {
  log.Fatal(err)
 }
 defer srcFile.Close()

 if _, err := srcFile.WriteTo(dstFile); err != nil {
  log.Fatal(err)
 }

 fmt.Println("Arquivo baixado com sucesso:", localFilePath)

 // Configuração do cliente S3
 sess := session.Must(session.NewSession(&aws.Config{
  Region: aws.String("us-west-2"),
 }))
 uploader := s3manager.NewUploader(sess)

 // Carregar arquivo para o S3
 file, err := os.Open(localFilePath)
 if err != nil {
  log.Fatal(err)
 }
 defer file.Close()

 _, err = uploader.Upload(&s3manager.UploadInput{
  Bucket: aws.String("seu-bucket-s3"),
  Key:    aws.String(filepath.Base(localFilePath)),
  Body:   file,
 })
 if err != nil {
  log.Fatal("Falha ao carregar arquivo para o S3:", err)
 }

 fmt.Println("Arquivo carregado com sucesso no S3")
}

2단계: Terraform 구성

Terraform은 AWS에서 Lambda 함수와 필수 리소스를 프로비저닝하는 데 사용됩니다. Lambda 함수, IAM 정책 및 S3 버킷을 생성하는 데 필요한 구성으로 main.tf 파일을 생성합니다.

provider "aws" {
  region = "us-east-1"
}

resource "aws_iam_role" "lambda_execution_role" {
  name = "lambda_execution_role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17",
    Statement = [
      {
        Action = "sts:AssumeRole",
        Effect = "Allow",
        Principal = {
          Service = "lambda.amazonaws.com"
        },
      },
    ]
  })
}

resource "aws_iam_policy" "lambda_policy" {
  name        = "lambda_policy"
  description = "A policy that allows a lambda function to access S3 and SFTP resources"

  policy = jsonencode({
    Version = "2012-10-17",
    Statement = [
      {
        Action = [
          "s3:ListBucket",
          "s3:GetObject",
          "s3:PutObject",
        ],
        Effect = "Allow",
        Resource = [
          "arn:aws:s3:::seu-bucket-s3",
          "arn:aws:s3:::seu-bucket-s3/*",
        ],
      },
    ]
  })
}

resource "aws_iam_role_policy_attachment" "lambda_policy_attachment" {
  role       = aws_iam_role.lambda_execution_role.name
  policy_arn = aws_iam_policy.lambda_policy.arn
}

resource "aws_lambda_function" "sftp_lambda" {
  function_name = "sftp_lambda_function"

  s3_bucket = "seu-bucket-s3-com-codigo-lambda"
  s3_key    = "sftp-lambda.zip"

  handler = "main"
  runtime = "go1.x"

  role = aws_iam_role.lambda_execution_role.arn

  environment {
    variables = {
      SFTP_HOST     = "endereco_sftp",
      SFTP_USER     = "seu_usuario_sftp",
      SFTP_PASSWORD = "sua_senha_sftp",
      S3_BUCKET     = "seu-bucket-s3",
    }
  }
}

resource "aws_s3_bucket" "s3_bucket" {
  bucket = "seu-bucket-s3"
  acl    = "private"
}

3단계: GitLab CI/CD 구성

GitLab에서 .gitlab-ci.yml 파일에 CI/CD 파이프라인을 정의합니다. 이 파이프라인에는 Go 애플리케이션을 테스트하는 단계, Terraform을 실행하여 인프라를 프로비저닝하는 단계, 필요한 경우 정리 단계가 포함되어야 합니다.

stages:
  - test
  - build
  - deploy

variables:
  S3_BUCKET: "seu-bucket-s3"
  AWS_DEFAULT_REGION: "us-east-1"
  TF_VERSION: "1.0.0"

before_script:
  - 'which ssh-agent || ( apt-get update -y && apt-get install openssh-client -y )'
  - eval $(ssh-agent -s)
  - echo "$PRIVATE_KEY" | tr -d '\r' | ssh-add -
  - mkdir -p ~/.ssh
  - chmod 700 ~/.ssh
  - ssh-keyscan -H 'endereco_sftp' >> ~/.ssh/known_hosts

test:
  stage: test
  image: golang:1.18
  script:
    - go test -v ./...

build:
  stage: build
  image: golang:1.18
  script:
    - go build -o myapp
    - zip -r sftp-lambda.zip myapp
  artifacts:
    paths:
      - sftp-lambda.zip
  only:
    - master

deploy:
  stage: deploy
  image: hashicorp/terraform:$TF_VERSION
  script:
    - terraform init
    - terraform apply -auto-approve
  only:
    - master
  environment:
    name: production

4단계: Databricks와 통합

파일을 S3에 업로드한 후 Lambda 함수는 Databricks에서 작업을 트리거해야 합니다. 이는 Databricks API를 사용하여 기존 작업을 시작할 수 있습니다.

package main

import (
 "bytes"
 "encoding/json"
 "fmt"
 "net/http"
)

// Estrutura para a requisição de iniciar um job no Databricks
type DatabricksJobRequest struct {
 JobID int `json:"job_id"`
}

// Função para acionar um job no Databricks
func triggerDatabricksJob(databricksInstance string, token string, jobID int) error {
 url := fmt.Sprintf("https://%s/api/2.0/jobs/run-now", databricksInstance)
 requestBody, _ := json.Marshal(DatabricksJobRequest{JobID: jobID})
 req, err := http.NewRequest("POST", url, bytes.NewBuffer(requestBody))
 if err != nil {
  return err
 }

 req.Header.Set("Content-Type", "application/json")
 req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))

 client := &http.Client{}
 resp, err := client.Do(req)
 if err != nil {
  return err
 }
 defer resp.Body.Close()

 if resp.StatusCode != http.StatusOK {
  return fmt.Errorf("Failed to trigger Databricks job, status code: %d", resp.StatusCode)
 }

 return nil
}

func main() {
 // ... (código existente para conectar ao SFTP e carregar no S3)

 // Substitua pelos seus valores reais
 databricksInstance := "your-databricks-instance"
 databricksToken := "your-databricks-token"
 databricksJobID := 123 // ID do job que você deseja acionar

 // Acionar o job no Databricks após o upload para o S3
 err := triggerDatabricksJob(databricksInstance, databricksToken, databricksJobID)
 if err != nil {
  log.Fatal("Erro ao acionar o job do Databricks:", err)
 }

 fmt.Println("Job do Databricks acionado com sucesso")
}

5단계: 파이프라인 실행

파이프라인을 실행할 GitLab 저장소에 코드를 푸시하세요. 모든 단계가 성공적으로 완료되었는지, Lambda 함수가 작동하고 S3 및 Databricks와 올바르게 상호 작용하는지 확인하십시오.

전체 코드와 .gitlab-ci.yml 파일이 구성되면 다음 단계에 따라 파이프라인을 실행할 수 있습니다.

• GitLab 저장소에 코드를 푸시하세요.

  git add .
  git commit -m "Adiciona função Lambda para integração SFTP, S3 e Databricks"
  git push origin master

git add .
git commit -m "Adiciona função Lambda para integração SFTP, S3 e Databricks"
git push origin master
´´´

• GitLab CI/CD는 새 커밋을 감지하고 자동으로 파이프라인을 시작합니다.
• 리포지토리의 CI/CD 섹션에 액세스하여 GitLab에서 파이프라인 실행을 추적하세요.
• 모든 단계가 성공하면 Lambda 함수가 배포되어 사용할 준비가 됩니다.

액세스 토큰 및 개인 키와 같은 민감한 정보를 저장하려면 GitLab CI/CD에서 환경 변수를 구성해야 한다는 점을 기억하세요. GitLab 프로젝트의 '설정' > 'CI/CD' > '변수' 섹션에서 수행할 수 있습니다.

또한 Databricks 토큰에 작업을 트리거하는 데 필요한 권한이 있고 해당 작업이 제공된 ID로 존재하는지 확인하세요.

결론

GitLab CI/CD, Terraform, AWS Lambda와 같은 도구를 사용하면 데이터 엔지니어링 작업 자동화를 크게 단순화할 수 있습니다. 이 문서에 설명된 단계를 따르면 Go의 효율성과 단순성을 모두 활용하여 SFTP, S3 및 Databricks 간의 데이터 수집 및 통합을 자동화하는 강력한 시스템을 만들 수 있습니다. 대규모 데이터 통합의 과제.

내 연락처:

LinkedIn - Airton Lira Junior

iMasters - 에어튼 리라 주니어

aws #lambda #terraform #gitlab #ci_cd #go #databricks #dataengineering #automation