0

理想的には、切り取った次のコードが機能します。 

import kudu 
from kudu.client import Partitioning

df = … #some spark dataframe 

# Connect to Kudu master server 
client = kudu.connect(host=‘…‘, port=7051)

# infer schema from spark dataframe
schema = df.schema 

# Define partitioning schema 
partitioning = Partitioning().add_hash_partitions(column_names=['key'], num_buckets=3) 

# Create new table 
client.create_table('dev.some_example', schema, partitioning)

ただし、 client.create_table は、データフレームからの構造体ではなく、kudu.schema.Schema を想定しています。ただし、Scala ではこれを行うことができます ( https://kudu.apache.org/docs/developing.htmlから):

kuduContext.createTable(
"dev.some_example", df.schema, Seq("key"),
new CreateTableOptions()
    .setNumReplicas(1)
    .addHashPartitions(List("key").asJava, 3))

kuduスキーマビルダーで各列を手動で定義せずに、PySparkで同じことができるかどうか疑問に思っていましたか?

4

1 に答える 1

0

そこで、PySpark データフレーム スキーマを kudu.schema.Schema に変換するヘルパー関数を自分で作成しました。これが誰かの役に立てば幸いです。フィードバックをお待ちしております。

補足として、データ型マッピングを追加または編集することをお勧めします。

import kudu
from kudu.client import Partitioning
def convert_to_kudu_schema(df_schema, primary_keys):
    builder = kudu.schema.SchemaBuilder()
    data_type_map = {
        "StringType":kudu.string,
        "LongType":kudu.int64,
        "IntegerType":kudu.int32,
        "FloatType":kudu.float,
        "DoubleType":kudu.double,
        "BooleanType":kudu.bool,
        "TimestampType":kudu.unixtime_micros,
    }

    for sf in df_schema:
        pk = False
        nullable=sf.nullable
        if (sf.name in primary_keys): 
            pk = True
            nullable = False

        builder.add_column(
            name=sf.name,
            nullable=nullable,
            type_=data_type_map[str(sf.dataType)]
        )
    builder.set_primary_keys(primary_keys)
    return builder.build()

次のように呼び出すことができます。

kudu_schema = convert_to_kudu_schema(df.schema,primary_keys=["key1","key2"])

私はまだよりエレガントなソリューションを求めています。;)

于 2018-10-31T10:43:38.140 に答える