如何通过pyspark将十六进制数据插入Cassandra中的blob数据类型列

如何解决如何通过pyspark将十六进制数据插入Cassandra中的blob数据类型列

我正在尝试将十六进制字符串插入具有blob数据类型列的Cassandra表中。 Cassandra表结构如下：

创建表mob.sample（ id文字PRIMARY KEY， 数据斑点 ）;

这是我的代码：

from pyspark.sql import SparkSession,SQLContext
from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark.sql.functions import udf


def hexstrtohexnum(hexstr):
    ani = int(hexstr[2:],16)
    return(ani)

# Create a DataFrame using SparkSession
spark = (SparkSession.builder
         .appName('SampleLoader')
         .appName('SparkCassandraApp')
         .getOrCreate())


schema = StructType([StructField("id",StringType(),True),StructField("data",True)])

# Create a DataFrame
df = spark.createDataFrame([("key1",'0x546869732069732061206669727374207265636f7264'),("key2",'0x546865207365636f6e64207265636f7264'),("key3",'0x546865207468697264207265636f7264')],schema)

hexstr2hexnum = udf(lambda z: hexstrtohexnum(z),IntegerType())
spark.udf.register("hexstr2hexnum",hexstr2hexnum)
df.withColumn("data",hexstr2hexnum("data"))
df.write.format("org.apache.spark.sql.cassandra").options(keyspace='mob',table='sample').save(mode="append")

当我运行上面的代码时，出现错误：

WARN  2020-09-03 19:41:57,902 org.apache.spark.scheduler.TaskSetManager: Lost task 3.0 in stage 17.0 (TID 441,10.37.122.156,executor 2): com.datastax.spark.connector.types.TypeConversionException: Cannot convert object 0x546869732069732061206669727374207265636f7264 of type class java.lang.String to java.nio.ByteBuffer.
    at com.datastax.spark.connector.types.TypeConverter$$anonfun$convert$1.apply(TypeConverter.scala:44)
    at com.datastax.spark.connector.types.TypeConverter$ByteBufferConverter$$anonfun$convertPF$11.applyOrElse(TypeConverter.scala:258)
    at com.datastax.spark.connector.types.TypeConverter$class.convert(TypeConverter.scala:42)
    at com.datastax.spark.connector.types.TypeConverter$ByteBufferConverter$.com$datastax$spark$connector$types$NullableTypeConverter$$super$convert(TypeConverter.scala:255)

这是数据框的内容。

>>> df.show(3)
+----+--------------------+
|  id|                data|
+----+--------------------+
|key1|0x546869732069732...|
|key2|0x546865207365636...|
|key3|0x546865207468697...|
+----+--------------------+

有人可以帮助我我的代码有什么问题吗？有什么我想念的吗？

解决方法

读取测试记录时，blob类型显示为BinaryType而不是StringType

>>> table1 = spark.read.format("org.apache.spark.sql.cassandra").options(table="blobtest",keyspace="test").load()
>>> table1.show()
+----+--------------------+
|  f1|                  f2|
+----+--------------------+
|1234|[54 68 69 73 20 6...|
+----+--------------------+

>>> print(table1.schema)
StructType(List(StructField(f1,StringType,false),StructField(f2,BinaryType,true)))

将架构更改为BinaryType，您应该可以编写

>>> string = "This is a test."
>>> arr = bytearray(string,'utf-8')
>>> schema = StructType([StructField("f1",StringType(),True),StructField("f2",BinaryType(),True)])
>>> df = spark.createDataFrame([("key3",arr)],schema)
>>> df.show()
        +----+--------------------+
        |  f1|                  f2|
        +----+--------------------+
        |key3|[54 68 69 73 20 6...|
        +----+--------------------+
        
>>> df.write.format("org.apache.spark.sql.cassandra").options(keyspace='test',table='blobtest2').save(mode="append")

,

我终于找到了解决此问题的方法。

interface IPojo {
  [key: string]: any,}

const checkType = (x: IPojo): x is Pet => {
  return 'name' in x && typeof (x as {
    name: unknown,}).name === 'string';
};

function foo(p: Pet) {
  console.log(p.name);
}

const bar: IPojo = {}
const baz = 'name';
bar[baz] = 'hi';

foo(bar); // TypeError,compiler can't verify bar.name
if (checkType(bar)) {
  foo(bar); // No TypeError,type narrowed correctly by your guard
}