如何在禁用AWS CLI的情况下通过PySpark连接AWS Glue作业与Amazon Keyspaces

当然可以实现，以下是两种无需依赖AWS CLI的方案，适配Glue作业的运行环境：

方案一：Spark Cassandra Connector（批量操作首选） #

适合大规模数据的读写操作，利用Spark的分布式处理能力：

• 配置作业依赖：在Glue作业的「作业参数」中添加以下参数，引入兼容的Spark Cassandra Connector（版本需匹配Glue的Spark版本，示例对应Glue 4.0/Spark 3.3）：

  --additional-jars com.datastax.spark:spark-cassandra-connector-assembly_2.12:3.4.1
  

• 编写PySpark代码：通过IAM认证连接Keyspaces（依赖Glue作业关联的IAM角色权限）：

  from pyspark.sql import SparkSession
  
  # 初始化Spark会话并配置Keyspaces连接参数
  spark = SparkSession.builder \
      .appName("KeyspacesGlueBatchJob") \
      .config("spark.cassandra.connection.host", "cassandra.us-east-1.amazonaws.com")  # 替换为你的Keyspaces区域端点
      .config("spark.cassandra.connection.port", "9142") \
      .config("spark.cassandra.auth.provider", "com.datastax.oss.driver.api.auth.auth.iam.DefaultAwsAuthProvider") \
      .config("spark.cassandra.connection.ssl.enabled", "true") \
      .getOrCreate()
  
  # 读取Keyspaces表数据
  df = spark.read \
      .format("org.apache.spark.sql.cassandra") \
      .options(table="your_target_table", keyspace="your_keyspace") \
      .load()
  
  # 写入数据到Keyspaces（示例为追加模式，可根据需求改为overwrite等）
  df.write \
      .format("org.apache.spark.sql.cassandra") \
      .options(table="your_target_table", keyspace="your_keyspace") \
      .mode("append") \
      .save()
  

• 权限配置：给Glue作业关联的IAM角色添加AmazonKeyspacesFullAccess（或更细粒度的自定义策略，指定具体keyspace和操作），无需CLI配置权限。

方案二：Cassandra Python驱动（小批量/单条操作） #

适合少量数据的查询、插入或更新操作：

• 配置作业依赖：在Glue作业的「作业参数」中添加以下参数，自动安装Cassandra Python驱动：

  --additional-python-modules cassandra-driver
  

• 编写PySpark代码：通过IAM认证直接连接Keyspaces：

  from cassandra.cluster import Cluster
  from cassandra.auth import AwsIamAuthProvider
  
  # 初始化集群连接
  cluster = Cluster(
      contact_points=['cassandra.us-east-1.amazonaws.com'],  # 替换为你的区域端点
      port=9142,
      ssl_context=True,
      auth_provider=AwsIamAuthProvider()
  )
  
  # 连接目标keyspace
  session = cluster.connect('your_keyspace')
  
  # 执行查询
  rows = session.execute("SELECT * FROM your_target_table LIMIT 10")
  for row in rows:
      print(row)
  
  # 执行插入
  session.execute(
      "INSERT INTO your_target_table (id, content) VALUES (%s, %s)",
      (1001, "sample_data")
  )
  
  # 关闭连接
  session.shutdown()
  cluster.shutdown()
  

关键注意事项 #

• 替换代码中的区域端点、keyspace名称和表名为你的实际资源；
• Glue作业的IAM角色必须拥有对应Keyspaces的操作权限，权限配置在IAM控制台完成即可，无需CLI；
• Spark Cassandra Connector版本需与Glue的Spark版本兼容，可参考官方文档确认对应版本。

内容的提问来源于stack exchange，提问作者Sai Teja