AWS Glue for SparkからDatabricksのテーブルにアクセスしてみた

.
├── CFn_resource.yml
└── glue_scripts
    └── sample.py

AWSTemplateFormatVersion: '2010-09-09'
Description: CFn Template for Glue Spark Job with Databricks Connection

Parameters:
  # Glue用パラメーター
  GlueScriptS3Bucket:
    Type: String
    Description: S3 bucket containing the Glue job script

  # Databricks接続用パラメータ
  DatabricksHost:
    Type: String
    Description: 'Databricks host (e.g. dbc-123abc45-6def.cloud.databricks.com)'

  HttpPath:
    Type: String
    Description: 'HTTP Path for the SQL Warehouse (e.g. sql/1.0/warehouses/abcd1234)'

  DatabricksToken:
    Type: String
    Description: 'Databricks personal access token'
    NoEcho: true

Metadata:
  AWS::CloudFormation::Interface:
    ParameterGroups:
      - Label:
          default: Parameter for Glue
        Parameters:
          - GlueScriptS3Bucket
      - Label:
          default: Parameter for Databricks connection
        Parameters:
          - DatabricksHost
          - HttpPath
          - DatabricksToken

Resources:
  # Databricks接続情報を保存するためのシークレット
  DatabricksSecret:
    Type: AWS::SecretsManager::Secret
    Properties:
      Name: !Sub 'databricks/connection'
      Description: 'Databricks connection information for Glue Job'
      SecretString: !Sub |
        {
          "host": "${DatabricksHost}",
          "httpPath": "${HttpPath}",
          "token": "${DatabricksToken}"
        }

  # Glueジョブ用のIAMロール
  GlueJobRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: uehara-test-glue-job-for-databricks-role
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: glue.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
      Policies:
        - PolicyName: uehara-test-glue-job-for-databricks-policy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - s3:GetObject
                  - s3:PutObject
                  - s3:ListBucket
                Resource:
                  - !Sub 'arn:aws:s3:::${GlueScriptS3Bucket}'
                  - !Sub 'arn:aws:s3:::${GlueScriptS3Bucket}/*'
              - Effect: Allow
                Action:
                  - secretsmanager:GetSecretValue
                Resource: !Ref DatabricksSecret

  # Glue Sparkジョブ
  GlueJob:
    Type: AWS::Glue::Job
    Properties:
      Name: uehara-test-glue-job-for-databricks
      Role: !GetAtt GlueJobRole.Arn
      Command:
        Name: glueetl
        ScriptLocation: !Sub 's3://${GlueScriptS3Bucket}/sample.py' # スクリプトファイルの格納先
        PythonVersion: 3
      DefaultArguments:
        '--job-language': 'python'
        '--enable-metrics': 'true'
        '--enable-continuous-cloudwatch-log': 'true'
        '--enable-spark-ui': 'true'
        '--spark-event-logs-path': !Sub 's3://${GlueScriptS3Bucket}/sparkHistoryLogs/'
        '--secret_id': !Ref DatabricksSecret
        '--extra-jars': !Sub 's3://${GlueScriptS3Bucket}/DatabricksJDBC42.jar' # JARファイルの格納先
      GlueVersion: 5.0
      WorkerType: G.1X #  1 DPU
      NumberOfWorkers: 2
      MaxRetries: 0
      Timeout: 60 # minutes

import json
import sys

import boto3
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from pyspark.sql import SparkSession

# ジョブパラメータを取得
args = getResolvedOptions(sys.argv, ['JOB_NAME', 'secret_id'])

# Spark/Glueコンテキストの初期化
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Secrets Managerから接続情報を取得
def get_secret(secret_id):
    client = boto3.client('secretsmanager')
    response = client.get_secret_value(SecretId=secret_id)
    secret = response['SecretString']
    return json.loads(secret)

# 接続情報を取得
secret_id = args['secret_id']
connection_info = get_secret(secret_id)

# 接続パラメータを設定
databricks_host = connection_info['host']
http_path = connection_info['httpPath']
token = connection_info['token']

print(f"Connecting to Databricks host: {databricks_host}")
print(f"Using HTTP path: {http_path}")

# Databricksからデータを読み込む
try:
    # JDBC URLを構築
    jdbc_url = f"jdbc:databricks://{databricks_host}:443;httpPath={http_path};AuthMech=3;UID=token;PWD={token};UseNativeQuery=0"

    # JDBCを使用してDatabricksに接続
    df = spark.read \
        .format("jdbc") \
        .option("driver", "com.databricks.client.jdbc.Driver") \
        .option("url", jdbc_url) \
        .option("dbtable", "workspace.test_managed_schema.my_sample_data") \
        .load()

    print(f"Successfully read {df.count()} rows from Databricks")

    # 取得したデータの表示
    df.show(10)

except Exception as e:
    print(f"Error connecting to Databricks: {str(e)}")
    raise

job.commit()

jdbc:databricks://{databricks_host}:443;httpPath={http_path};AuthMech=3;UID=token;PWD={token}

An error occurred while calling o146.showString. [Databricks][JDBC](10140) Error converting value to int.

jdbc:databricks://{databricks_host}:443;httpPath={http_path};AuthMech=3;UID=token;PWD={token};UseNativeQuery=0