Я пытаюсь написать данные в Delta Table без использования Spark. Таким образом, я изучаю API Delta ядра, чтобы написать данные в Delta Tables.
Когда я пытаюсь написать данные, которые разделены на более чем 1 столбец. Я получаю ошибку как
java.lang.runtimeexception: java.lang.unsupportedoperationexception: org.apache.parquet.column.values.dictionary.dictionaryvalueswriter $ plaintegerdictionar несоответствие. Однако те же данные записываются в таблицу дельты, если таблица разделена на 1 одно столбец. < /P>
public void insertData(String tablePath, DeltaApiConfiguration deltaApiConfiguration) throws IOException {
Configuration hadoopConf = new Configuration();
Engine engine = DefaultEngine.create(hadoopConf);
hadoopConf.set("fs.defaultFS", deltaApiConfiguration.getDefaultFS());
Table table = Table.forPath(engine, tablePath);
List dataActions = new ArrayList();
TransactionBuilder txnBuilder =
table.createTransactionBuilder(
engine,
"Insert data to delta table", /* engineInfo */
Operation.MANUAL_UPDATE);
Transaction txn = txnBuilder.build(engine);
List partitionedColumns = txn.getPartitionColumns(engine);
Row txnState = txn.getTransactionState(engine);
StructType schema = txn.getSchema(engine);
FilteredColumnarBatch batch1 = generatedPartitionedDataBatch(5 ,schema, partitionedColumns);
CloseableIterator data = toCloseableIterator(Collections.singletonList(batch1).iterator());
Map partitionValues = Map.of(
"STATE", Literal.ofString("Goa"),
"CITY", Literal.ofString("Vasco")
);
CloseableIterator partitionedDataActions = getPartitionedData(engine, txnState, data, partitionValues);
while (partitionDataActions.hasNext()) {
System.out.println("while loop!!!");
dataActions.add(partitionDataActions.next());
}
CloseableIterable dataActionsIterable = CloseableIterable.inMemoryIterable(toCloseableIterator(partitionedDataActions));
txn.commit(engine, dataActionsIterable);
}
public static CloseableIterator getPartitionedData(Engine engine, Row txnState, CloseableIterator data, Map partitionValues) throws IOException {
CloseableIterator dataFiles = null;
DataWriteContext writeContext;
try {
CloseableIterator physicalData = Transaction.transformLogicalData(engine, txnState, data, partitionValues);
writeContext = Transaction.getWriteContext(engine, txnState, partitionValues);
dataFiles = engine.getParquetHandler().writeParquetFiles(writeContext.getTargetDirectory(), physicalData, writeContext.getStatisticsColumns());
} catch (Exception e) {
e.printStackTrace();
throw new IOException("Error while writing data to delta table");
} finally {
if (data != null) {
data.close();
}
}
return Transaction.generateAppendActions(engine, txnState, dataFiles, writeContext);
}
< /code>
Я получаю эту ошибку в цикле, когда значения распакованы. Для данных, разделенных на одном столбце, этот код работает нормально.
Я использую ниже ссылку для изучения таблиц Delta.>
Подробнее здесь: https://stackoverflow.com/questions/796 ... kernel-api
