Kosten senken mit Prozessautomatisierung in Databricks

Ich hatte bei einem Kunden das Bedürfnis, die Kosten für Prozesse zu senken, die auf Databricks ausgeführt wurden. Eine der Funktionen, für die Databricks verantwortlich war, war das Sammeln von Dateien aus verschiedenen SFTP, deren Dekomprimierung und deren Platzierung im Data Lake.

Die Automatisierung von Datenworkflows ist eine entscheidende Komponente im modernen Data Engineering. In diesem Artikel erfahren Sie, wie Sie mit GitLab CI/CD und Terraform eine AWS Lambda-Funktion erstellen, die es einer Go-Anwendung ermöglicht, eine Verbindung zu einem SFTP-Server herzustellen, Dateien zu sammeln, sie in Amazon S3 zu speichern und schließlich einen Job auf Databricks auszulösen. Dieser End-to-End-Prozess ist für Systeme, die auf effiziente Datenintegration und Automatisierung angewiesen sind, unerlässlich.

Was Sie für diesen Artikel benötigen

• GitLab-Konto mit einem Repository für das Projekt.
• AWS-Konto mit Berechtigungen zum Erstellen von Lambda-, S3- und IAM-Ressourcen.
• Databricks-Konto mit Berechtigungen zum Erstellen und Ausführen von Jobs.
• Grundkenntnisse in Go, Terraform und GitLab CI/CD.

Schritt 1: Vorbereiten der Go-Anwendung

Erstellen Sie zunächst eine Go-Anwendung, die eine Verbindung zum SFTP-Server herstellt, um Dateien zu sammeln. Verwenden Sie Pakete wie github.com/pkg/sftp, um die SFTP-Verbindung herzustellen, und github.com/aws/aws-sdk-go, um mit dem AWS S3-Dienst zu interagieren.

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")
}

Schritt 2: Terraform konfigurieren

Terraform wird zur Bereitstellung der Lambda-Funktion und der erforderlichen Ressourcen auf AWS verwendet. Erstellen Sie eine main.tf-Datei mit der Konfiguration, die zum Erstellen der Lambda-Funktion, IAM-Richtlinien und S3-Buckets erforderlich ist.

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"
}

Schritt 3: GitLab CI/CD konfigurieren

Definieren Sie in GitLab die CI/CD-Pipeline in der Datei .gitlab-ci.yml. Diese Pipeline sollte Schritte zum Testen der Go-Anwendung, zum Ausführen von Terraform zur Bereitstellung der Infrastruktur und gegebenenfalls einen Schritt zur Bereinigung enthalten.

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

Schritt 4: Integration mit Databricks

Nach dem Hochladen der Dateien in S3 muss die Lambda-Funktion einen Job in Databricks auslösen. Dies kann mithilfe der Databricks-API erfolgen, um vorhandene Jobs zu starten.

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")
}

Schritt 5: Ausführen der Pipeline

Übertragen Sie den Code in das GitLab-Repository, damit die Pipeline ausgeführt werden kann. Stellen Sie sicher, dass alle Schritte erfolgreich abgeschlossen wurden und dass die Lambda-Funktion betriebsbereit ist und korrekt mit S3 und Databricks interagiert.

Sobald Sie den vollständigen Code und die Datei .gitlab-ci.yml konfiguriert haben, können Sie die Pipeline ausführen, indem Sie die folgenden Schritte ausführen:

• Schieben Sie Ihren Code in das GitLab-Repository:

  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 erkennt den neuen Commit und startet die Pipeline automatisch.
• Verfolgen Sie die Ausführung der Pipeline in GitLab, indem Sie auf den CI/CD-Bereich Ihres Repositorys zugreifen.
• Wenn alle Phasen erfolgreich sind, wird Ihre Lambda-Funktion bereitgestellt und ist einsatzbereit.

Denken Sie daran, dass Sie Umgebungsvariablen in GitLab CI/CD konfigurieren müssen, um vertrauliche Informationen wie Zugriffstokens und private Schlüssel zu speichern. Dies kann im Abschnitt „Einstellungen“ > „CI/CD“ > „Variablen“ Ihres GitLab-Projekts erfolgen.

Stellen Sie außerdem sicher, dass das Databricks-Token über die erforderlichen Berechtigungen zum Auslösen von Aufträgen verfügt und dass der Auftrag mit der angegebenen ID vorhanden ist.

Abschluss

Die Automatisierung von Data-Engineering-Aufgaben kann mit Tools wie GitLab CI/CD, Terraform und AWS Lambda erheblich vereinfacht werden. Indem Sie die in diesem Artikel beschriebenen Schritte befolgen, können Sie ein robustes System erstellen, das die Datenerfassung und Integration zwischen SFTP, S3 und Databricks automatisiert, und das alles mit der Effizienz und Einfachheit von Go. Mit diesem Ansatz sind Sie gut gerüstet, um das Problem anzugehen Herausforderungen der Datenintegration im großen Maßstab.

Meine Kontakte:

LinkedIn – Airton Lira Junior

iMasters - Airton Lira Junior

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