【发布时间】：2020-02-25 08:48:19
【问题描述】：
我正在尝试使用摄取 sdk 在我的 Java 程序中摄取文件并收到错误。我已经提前创建了表格、舞台和管道。我还将我的 CSV 文件放入内部阶段以准备摄取。现在我的 Java 程序只需要调用 ingestFiles 来加载文件。但是，当我拨打电话时，我收到 404 错误（请参阅下面的附加日志文件和 Java 程序）。
我已使用以下程序创建了我的公钥和私钥： https://docs.snowflake.net/manuals/user-guide/data-load-snowpipe-rest-gs.html#step-3-configure-security-per-user
另外请注意，我可以通过手动发出与管道关联的 COPY 命令来成功加载文件。不过，我更喜欢使用摄取 SDK，以便更好地捕获响应。
我怀疑这是权限问题，但不确定。任何帮助将不胜感激。
03:11:06.275 [main] INFO com.pardi.snowpipetest.IngestTest - 从文件 keys/rsa_key.p8 成功加载私钥
[main] WARN net.snowflake.ingest.connection.RequestBuilder - 无法读取版本信息：java.nio.file.FileSystemNotFoundException
[main] INFO net.snowflake.ingest.connection.SecurityManager - 使用主题 WX11111.JPARDI 创建 Token
[main] INFO net.snowflake.ingest.connection.SecurityManager - 使用颁发者 WX11111.JPARDI.SHA256:xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx 创建令牌
[main] INFO net.snowflake.ingest.connection.SecurityManager - 创建新的 JWT - xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
[main] INFO net.snowflake.ingest.connection.RequestBuilder - 使用参数创建 RequestBuilder：帐户：WX11111，用户：JPARDI，方案：https，主机：.us-east-1.snowflakecomputing.com，端口： 443
[main] INFO net.snowflake.ingest.SimpleIngestManager - 发送请求 UUID -
[main] INFO net.snowflake.ingest.connection.RequestBuilder - 创建插入请求：https://.us-east-1.snowflakecomputing.com:443/v1/data/pipes/DEMO_DB.PUBLIC.COVETRUS_SNOWFLAKE_BATCH_SINK_SNOWPIPETEST_PIPE_H_CLIENT_0/ insertFiles?requestId=39a2c1e4-c637-4424-bb68-42c4eb71873a
[main] INFO net.snowflake.ingest.SimpleIngestManager - 尝试解组插入响应 - HttpResponseProxy{HTTP/1.1 404 Not Found [Content-Type: application/json, Date: Tue, 25 Feb 2020 08:11:10 GMT，服务器：nginx，Strict-Transport-Security：max-age=31536000，X-Content-Type-Options：nosniff，X-Frame-Options：拒绝，连接：keep-alive] net.snowflake.ingest.internal。 apache.http.client.entity.DecompressingEntity@29c80149}
[main] WARN net.snowflake.ingest.connection.ServiceResponseHandler - 在 unmarshallInsert 响应中发现异常状态代码 - 404
[main] 错误 net.snowflake.ingest.connection.ServiceResponseHandler - 在服务响应中发现状态码 404
03:11:11.943 [main] INFO com.pardi.snowpipetest.IngestTest - 服务异常：
HTTP 状态：404
{
消息：指定的对象不存在或未授权。未找到管道，
数据：空
}
package com.pardi.snowpipetest;
import net.snowflake.ingest.SimpleIngestManager;
import net.snowflake.ingest.connection.HistoryResponse;
import net.snowflake.ingest.connection.IngestResponse;
import net.snowflake.ingest.connection.IngestResponseException;
import org.bouncycastle.asn1.pkcs.PrivateKeyInfo;
import org.bouncycastle.jcajce.provider.BouncyCastleFipsProvider;
import org.bouncycastle.openssl.PEMParser;
import org.bouncycastle.openssl.jcajce.JcaPEMKeyConverter;
import org.bouncycastle.openssl.jcajce.JceOpenSSLPKCS8DecryptorProviderBuilder;
import org.bouncycastle.operator.InputDecryptorProvider;
import org.bouncycastle.pkcs.PKCS8EncryptedPrivateKeyInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.ClassPathResource;
import java.io.*;
import java.security.PrivateKey;
import java.security.Security;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.*;
public class IngestTest {
private static final Logger LOGGER = LoggerFactory.getLogger(IngestTest.class);
private static String host = "<xxxx>.us-east-1.snowflakecomputing.com";
private static String account = "<xxxxx>";
private static String user = "xxxxxx";
private static String passPhrase = "xxxxxxxxx";
private static int port = 443;
private static String database = "DEMO_DB";
private static String schema = "PUBLIC";
private static String pipe = "COVETRUS_SNOWFLAKE_BATCH_SINK_SNOWPIPETEST_PIPE_H_CLIENT_0";
private static String fqPipe = database + "." + schema + "." + pipe;
private static PrivateKey privateKey;
private static SimpleIngestManager manager;
public static void main(String[] args) throws Exception {
String privateKeyStr = loadPrivateKey();
privateKey = parseEncryptedPrivateKey(privateKeyStr, passPhrase);
manager = new SimpleIngestManager(account, user, fqPipe, privateKey, "https", host, port);
Set<String> files = new TreeSet<>();
files.add("covetrus_snowflake_batch_sink_snowpipetest_stage_h_client/h_client.csv");
try {
IngestResponse response = manager.ingestFiles(manager.wrapFilepaths(files), null);
LOGGER.info("response=" + response.toString());
HistoryResponse history = waitForFilesHistory(files);
LOGGER.info("Received history response: " + history.toString());
} catch (IngestResponseException e) {
LOGGER.info("Service exception: " + e.toString());
} catch (Exception e) {
LOGGER.info("Exception: " + e.getMessage());
}
}
private static String loadPrivateKey() throws IOException {
byte[] keyBytes;
String filename = "keys/rsa_key.p8";
File privateKeyFile = null;
try {
privateKeyFile = new ClassPathResource(filename).getFile();
FileInputStream fis = new FileInputStream(privateKeyFile);
DataInputStream dis = new DataInputStream(fis);
keyBytes = new byte[(int) privateKeyFile.length()];
dis.readFully(keyBytes);
dis.close();
} catch (IOException e) {
LOGGER.info("FATAL: error loading private key from file " + filename + ", exception=" + e.getMessage());
e.printStackTrace();
throw e;
}
String privateKeyStr = new String(keyBytes);
LOGGER.info("successfully loaded private key from file " + filename);
return privateKeyStr;
}
public static PrivateKey parseEncryptedPrivateKey(String key, String passphrase) {
Security.addProvider(new BouncyCastleFipsProvider());
try {
PEMParser pemParser = new PEMParser(new StringReader(key));
PKCS8EncryptedPrivateKeyInfo encryptedPrivateKeyInfo = (PKCS8EncryptedPrivateKeyInfo) pemParser.readObject();
pemParser.close();
InputDecryptorProvider pkcs8Prov = new JceOpenSSLPKCS8DecryptorProviderBuilder().build(passphrase.toCharArray());
JcaPEMKeyConverter converter = new JcaPEMKeyConverter().setProvider(BouncyCastleFipsProvider.PROVIDER_NAME);
PrivateKeyInfo decryptedPrivateKeyInfo = encryptedPrivateKeyInfo.decryptPrivateKeyInfo(pkcs8Prov);
return converter.getPrivateKey(decryptedPrivateKeyInfo);
} catch (Exception e) {
throw new RuntimeException("Invalid encrypted private key or passphrase");
}
}
private static HistoryResponse waitForFilesHistory(Set<String> files)
throws Exception {
ExecutorService service = Executors.newSingleThreadExecutor();
class GetHistory implements
Callable<HistoryResponse> {
private Set<String> filesWatchList;
GetHistory(Set<String> files) {
this.filesWatchList = files;
}
String beginMark = null;
public HistoryResponse call()
throws Exception {
HistoryResponse filesHistory = null;
while (true) {
Thread.sleep(500);
HistoryResponse response = manager.getHistory(null, null, beginMark);
if (response.getNextBeginMark() != null) {
beginMark = response.getNextBeginMark();
}
if (response != null && response.files != null) {
for (HistoryResponse.FileEntry entry : response.files) {
//if we have a complete file that we've
// loaded with the same name..
String filename = entry.getPath();
if (entry.getPath() != null && entry.isComplete() &&
filesWatchList.contains(filename)) {
if (filesHistory == null) {
filesHistory = new HistoryResponse();
filesHistory.setPipe(response.getPipe());
}
filesHistory.files.add(entry);
filesWatchList.remove(filename);
//we can return true!
if (filesWatchList.isEmpty()) {
return filesHistory;
}
}
}
}
}
}
}
GetHistory historyCaller = new GetHistory(files);
//fork off waiting for a load to the service
Future<HistoryResponse> result = service.submit(historyCaller);
HistoryResponse response = result.get(2, TimeUnit.MINUTES);
return response;
}
}
【问题讨论】：
标签： snowflake-cloud-data-platform