小编Nak*_*euh的帖子

我有这个代码在scala中运行良好:

val schema = StructType(Array(
        StructField("field1", StringType, true),
        StructField("field2", TimestampType, true),
        StructField("field3", DoubleType, true),
        StructField("field4", StringType, true),
        StructField("field5", StringType, true)
    ))

val df = spark.read
    // some options
    .schema(schema)
    .load(myEndpoint)

我想在Java中做类似的事情.所以我的代码如下:

final StructType schema = new StructType(new StructField[] {
     new StructField("field1",  new StringType(), true,new Metadata()),
     new StructField("field2", new TimestampType(), true,new Metadata()),
     new StructField("field3", new StringType(), true,new Metadata()),
     new StructField("field4", new StringType(), true,new Metadata()),
     new StructField("field5", new StringType(), true,new Metadata())
});

Dataset<Row> df = spark.read()
    // some options
    .schema(schema)
    .load(myEndpoint);

但这给了我以下错误: …

使用 flink SQL API，我想将多个表连接在一起并在时间窗口内进行一些计算。我有 3 个来自 CSV 文件的表，一个来自 Kafka。在 Kafka 表中，我有一个字段timestampMs，我想将其用于我的时间窗口操作。

为此，我做了以下代码：

reamExecutionEnvironment env = ... ;
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

TableSource table1 = CsvTableSource.builder()
        .path("path/to/file1.csv")
        .ignoreFirstLine()
        .fieldDelimiter(",")
        .field("id1", Types.STRING)
        .field("someInfo1", Types.FLOAT)
        .build();

TableSource table2 = CsvTableSource.builder()
        .path("path/to/file2.csv")
        .ignoreFirstLine()
        .fieldDelimiter(",")
        .field("id2", Types.STRING)
        .field("someInfo2", Types.STRING)
        .build();

TableSource table3 = CsvTableSource.builder()
        .path("path/to/file3.csv")
        .ignoreFirstLine()
        .fieldDelimiter(",")
        .field("id2", Types.STRING)
        .field("id1", Types.STRING)
        .field("someInfo3", Types.FLOAT)
        .build();

tableEnv.registerTableSource("Table1",table1);
tableEnv.registerTableSource("Table2",table2);
tableEnv.registerTableSource("Table3",table3);


Schema schemaExt = new Schema().schema(SOME_SCHEMA);
schemaExt = schemaExt.field("rowtime", Types.SQL_TIMESTAMP).rowtime(new Rowtime().timestampsFromField("timestampMs").watermarksPeriodicBounded(40000));

tableEnv.connect(new Kafka()
        .version("universal")
        .topic(MY_TOPIC)
        .properties(MY_PROPERTIES)
        .sinkPartitionerRoundRobin() …

我有一个包含2列tag和的数据框value。

我想补充一点，包含新列max的value列。（对于每行它将是相同的值）。

我尝试执行以下操作，但是没有成功。

val df2 = df.withColumn("max",max($"value"))

如何将max列添加到数据集？

我有两个数据帧df1，df2它们具有以下结构：

print(df1)
+-------+------------+-------------+---------+
|     id|      vector|  start_time | end_time|
+-------+------------+-------------+---------+
|      1| [0,0,0,0,0]|          000|      200|
|      2| [1,1,1,1,1]|          200|      500|
|      3| [0,1,0,1,0]|          100|      500|
+-------+------------+-------------+---------+

print(df2)
+-------+------------+-------+
|     id|      vector|   time|
+-------+------------+-------+
|      A| [0,1,1,1,0]|    050|
|      B| [1,0,0,1,1]|    150|
|      C| [1,1,1,1,1]|    250|
|      D| [1,0,1,0,1]|    350|
|      E| [1,1,1,1,1]|    450|
|      F| [1,0,5,0,0]|    550|
+-------+------------+-------+

我要的是：对于每一个数据df1，得到的所有数据来自df2于该time之间start_time，并end_time与所有这些数据计算两个向量之间的欧氏距离。

我从以下代码开始，但我在计算距离的过程中遇到了困难：

val joined_DF = kafka_DF.crossJoin(
        hdfs_DF.withColumnRenamed("id","id2").withColumnRenamed("vector","vector2")
    ) …

我正在使用 Flink SQL API，我在所有“模式”类型之间有点迷失：TableSchema、Schema（来自org.apache.flink.table.descriptors.Schema）和TypeInformation。

ATableSchema可以从 a 创建TypeInformation，aTypeInformation可以从 a 创建TableSchema，aSchema可以从 a 创建TableSchema

但看起来 aSchema无法转换回TypeInformationor TableSchema(？)

为什么有 3 种不同类型的对象来存储同一种信息？

例如，假设我有一个来自 Avro 架构文件的字符串架构，并且我想向其中添加一个新字段。为此，我找到的唯一解决方案是：

String mySchemaRaw = ...;
TypeInformation<Row> typeInfo = AvroSchemaConverter.convertToTypeInfo(mySchemaRaw);
Schema newSchema = new Schema().schema(TableSchema.fromTypeInfo(typeInfo));
newSchema = newSchema.field("nexField",...);


// Need the newSchema as a TableSchema 

这是使用这些对象的正常方式吗？（我觉得很奇怪）

我想从 DynamoDB 表存储和检索数据。

我的数据（项目=用户对应用程序功能的评论）具有以下属性：

user        string 
feature     string
appVersion  string
timestamp   string 
rate        int   
description string 

该应用程序的多个版本具有多种功能，并且用户可以对这些功能进行多次评论。所以我想使用 ( user, appVersion, feature, timestamp) 作为主键。

但似乎不可能在 DynamoDB 的主键中使用那么多属性。

我实现的第一个解决方案是使用useras a ，并将 ( , , )Partition Key的哈希值用作 a （在名为 的新字段中）。appVersionfeaturetimestampSort KeyreviewID

我的问题是，我想在不知道值的情况下检索给定user,的项目feature（假设我想要最新的项目，或与 3 个字段匹配的所有项目的列表）appVersiontimestamptimestamp

如果不知道timestamp，我就无法构建Sort Key检索我的物品所需的必要条件。timestamp但如果我从 中删除Sort Key，我将无法存储具有相同 ( user, appVersion, feature) …