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

【问题标题】：How to insert hex data into blob data type column in Cassandra via pyspark如何通过pyspark将十六进制数据插入Cassandra中的blob数据类型列
【发布时间】：2020-12-23 02:48:07
【问题描述】：

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

创建表 mob.sample ( id 文本主键，数据块 );

这是我的代码：

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",StringType(),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...|
+----+--------------------+

有人可以帮我看看我的代码有什么问题吗？有什么我遗漏的吗？

【问题讨论】：

标签： apache-spark pyspark cassandra blob spark-cassandra-connector

【解决方案1】：

读取测试记录，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")

【讨论】：

该解决方案仅在字符串是 ascii 字符串时才有效。但是，当字符串是十六进制 blob 字符串并且您希望将十六进制 blob 字符串按原样存储到 blob 列中时，它不起作用。是否有可能将十六进制 blob 字符串按原样存储到 blob 列中？
使用 bytearray.fromhex 将十六进制字符串转换为 bytearray 例如 df = spark.createDataFrame([("key4",bytearray.fromhex("546869732069732061206669727374207265636f7264"))],schema)

【解决方案2】：

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

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

def hextobinary(hexstr):
    a = binascii.unhexlify(hexstr[2:])
    return(a)

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


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

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

tobinary = udf(lambda z: hextobinary(z),BinaryType())
spark.udf.register("tobinary", tobinary)
df1 = df.withColumn("data",tobinary("data"))
print(df1)
df1.write.format("org.apache.spark.sql.cassandra").options(keyspace='mob',table='sample').save(mode="append")

【讨论】：