将数据从 Amazon redshift 卸载到 Amazon s3

Posted 2023-03-31

【中文标题】将数据从 Amazon redshift 卸载到 Amazon s3

【英文标题】：Unloading data from Amazon redshift to Amazon s3

【发布时间】：2018-09-02 15:21:55

【问题描述】：

我正在尝试使用以下代码将数据卸载到 S3 存储桶中。哪个有效，但卸载后会引发一些错误。

 Properties props = new Properties();
 props.setProperty("user", MasterUsername);
 props.setProperty("password", MasterUserPassword);
 conn = DriverManager.getConnection(dbURL, props);
 stmt = conn.createStatement(); 
 String sql;
 sql = "unload('select * from part where p_partkey in (select p_partkey from 
       part limit 10)') to"
       + " 's3://redshiftdump.****' "
       + " DELIMITER AS ','"
       + "ADDQUOTES " 
       + "NULL AS ''"
       + "credentials 'aws_access_key_id=****;aws_secret_access_key=***' "
       + "parallel off" + 
       ";"; 
 boolean i = stmt.execute(sql);
 stmt.close();
 conn.close();

卸货工作。它正在桶中创建一个文件。但它给了我一些错误

   java.sql.SQLException: 
      dataengine.impl.DSISimpleRowCountResult cannot be cast to 
      com.amazon.dsi.dataengine.interfaces.IResultSet
   at 
   com.amazon.redshift.core.jdbc42.PGJDBC42Statement.createResultSet(Unknown 
   Source)
   at com.amazon.jdbc.common.SStatement.executeQuery(Unknown Source)

这个错误是什么以及如何避免它？有没有办法以 CSV 格式转储表格。现在它正在以 FILE 格式转储文件。

【问题讨论】：

我得到了这个工作。使用 executeupdate 执行查询。但它是以 FILE 格式而不是 CSV 格式上传文件。

“文件格式”是什么意思？字段是否用逗号分隔？

您将文件命名为什么？是不是像 myfilename.csv 这样的东西？

@theDbGuy 是的，会是 myfilename.csv 吗？

不要在末尾添加.csv。它可以正常工作

【参考方案1】：

您说 UNLOAD 有效，但您收到此错误，这表明您连接成功，但查询完成时您的代码与 JDBC 驱动程序交互的方式存在问题。

我们在"Connect to Your Cluster Programmatically" 页面上提供了一个可能对我们的文档有所帮助的示例

关于输出文件格式，您将获得 UNLOAD SQL 中指定的任何内容，但文件名将有一个后缀（例如“000”或“6411_part_00”）以指示它是 UNLOAD 的哪一部分。

【讨论】：

您能否澄清您所说的“（例如“000”或“6411_part_00”）以指示它是UNLOAD 的哪一部分。”？我也得到不同的后缀，不确定它们的含义。也找不到任何文档。

【参考方案2】：

use executeUpdate .

  def runQuery(sql: String)  = 
    Class.forName("com.amazon.redshift.jdbc.Driver")
    val connection = DriverManager.getConnection(url, username, password)
    var statement: Statement = null
    try 
      statement = connection.createStatement()
      statement.setQueryTimeout(redshiftTimeoutInSeconds)
      val result = statement.executeUpdate(sql)
      logger.info(s"statement response code : $result")
     catch 
      case e: Exception => 
        logger.error(s"statement.isCloseOnCompletion :$e.getMessage ::: $e.printStackTrace()")
        throw new IngestionException(e.getMessage)
      
    
    finally 
      if(statement != null ) statement.close()
      connection.close()