I had a need at a client to reduce the cost of processes that ran on Databricks. One of the features that Databricks was responsible for was collecting files from various SFTP, decompressing them and placing them in the Data Lake.
Automating data workflows is a crucial component in modern data engineering. In this article, we will explore how to create an AWS Lambda function using GitLab CI/CD and Terraform that allows a Go application to connect to an SFTP server, collect files, store them in Amazon S3, and finally trigger a job on Databricks. This end-to-end process is essential for systems that rely on efficient data integration and automation.
Start by creating a Go application that will connect to the SFTP server to collect files. Use packages like github.com/pkg/sftp to establish the SFTP connection and github.com/aws/aws-sdk-go to interact with the AWS S3 service.
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")
}
Terraform will be used to provision the Lambda function and required resources on AWS. Create a main.tf file with the configuration required to create the Lambda function, IAM policies, and S3 buckets.
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"
}
In GitLab, define the CI/CD pipeline in the .gitlab-ci.yml file. This pipeline should include steps to test the Go application, run Terraform to provision the infrastructure, and a step for cleanup if necessary.
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
After uploading the files to S3, the Lambda function must trigger a job in Databricks. This can be done using the Databricks API to launch existing jobs.
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")
}
Push the code to the GitLab repository for the pipeline to run. Check that all steps are completed successfully and that the Lambda function is operational and interacting correctly with S3 and Databricks.
Once you have the complete code and the .gitlab-ci.yml file configured, you can run the pipeline by following these steps:
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")
}
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"
}
Remember that you will need to configure environment variables in GitLab CI/CD to store sensitive information such as access tokens and private keys. This can be done in the ‘Settings’ > ‘CI/CD’ > ‘Variables’ from your GitLab project.
Also, ensure that the Databricks token has the necessary permissions to trigger jobs and that the job exists with the provided ID.
Automating data engineering tasks can be significantly simplified using tools like GitLab CI/CD, Terraform, and AWS Lambda. By following the steps outlined in this article, you can create a robust system that automates data collection and integration between SFTP, S3, and Databricks, all with the efficiency and simplicity of Go. With this approach, you will be well equipped to address the challenges of data integration at scale.
My contacts:
LinkedIn - Airton Lira Junior
iMasters - Airton Lira Junior
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
The above is the detailed content of Implementing a Lambda with GitLab CI/CD and Terraform for SFTP Integration, S Databricks in Go. For more information, please follow other related articles on the PHP Chinese website!
![[Web front-end] Node.js quick start](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
