小编maz*_*cha的帖子

我使用以下命令来打开、替换、查看更改并保存文件：

sed 's/old string/new string/g' filename > filename1; diff filename1 filename; mv filename1 filename

是否可以在执行命令之前要求确认mv，如下所示？

sed 's/old string/new string/g' filename > filename1
diff filename1 filename

<Some-Command here for user to provide Yes or NO confirmation>

mv filename1 filename

目的是验证更改然后保存它。

我使用 Sparksessions read().json 方法来读取 json 文件，然后将其转换为镶木地板文件，它工作正常，但 .json(JAVARDD) 方法显示为已弃用的方法。我们可以有替代方法吗？当前使用java：spark版本2.4.3

我已经浏览了 Spark 库的文档，但没有得到任何替代方法

JavaSparkContext sc1 = JavaSparkContext.fromSparkContext(SparkContext.getOrCreate());
JavaRDD<String> rddData = sc1.parallelize(data);
Dataset<Row> dataDF = spark.read().json(rddData);

这里 .json 方法显示为已弃用的方法？我们可以有替代方法吗？

我已经完成了 How to parse JSON in Spark with fastxml without SparkSQL?

在这里他们建议使用 SqlContext 方法，该方法也已被弃用。

需要知道java Spark.read().json(JAVARDD对象)中.json的替代方法

我正在尝试创建一个 Spark 应用程序，它对于创建、读取、写入和更新 MySQL 数据很有用。那么，有没有办法使用Spark创建MySQL表呢？

下面我有一个 Scala-JDBC 代码，它在 MySQL 数据库中创建一个表。我怎样才能通过 Spark 做到这一点？

package SparkMysqlJdbcConnectivity

import org.apache.spark.sql.SparkSession
import java.util.Properties
import java.lang.Class
import java.sql.Connection
import java.sql.DriverManager

object MysqlSparkJdbcProgram {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("MysqlJDBC Connections")
      .master("local[*]")
      .getOrCreate()

    val driver = "com.mysql.jdbc.Driver"
    val url = "jdbc:mysql://localhost:3306/world"
    val operationtype = "create table"
    val tablename = "country"
    val tablename2 = "state"

    val connectionProperties = new Properties()

    connectionProperties.put("user", "root")
    connectionProperties.put("password", "root")

    val jdbcDf = spark.read.jdbc(url, s"${tablename}", connectionProperties)

    operationtype.trim() match { …

我正在尝试通过 Spark 结构化流从 Kafka 读取数据。但是，在Spark 2.4.0中，您无法为流设置group id（请参阅如何在Structured Streaming中为kafka数据源中的消费者组设置group.id？）。

然而，由于没有设置，spark 只是生成组 Id，而我陷入了 GroupAuthorizationException：

19/12/10 15:15:00 ERROR streaming.MicroBatchExecution: Query [id = 747090ff-120f-4a6d-b20e-634eb77ac7b8, runId = 63aa4cce-ad72-47f2-80f6-e87947b69685] terminated with error
org.apache.kafka.common.errors.GroupAuthorizationException: Not authorized to access group: spark-kafka-source-d2420426-13d5-4bda-ad21-7d8e43ebf518-1874352823-driver-2

请问有什么想法可以绕过这个吗？有趣的是，我可以通过 kafka-console-consumer.sh 读取这些数据，我可以在 .properties 文件中传递组 ID。

抛出异常的代码：

val df = spark
  .readStream
  .format("kafka")
  .option("subscribe", "topic")
  .option("startingOffsets", "earliest")
  .option("kafka.group.id", "idThatShouldBeUsed")
  .option("kafka.bootstrap.servers", "server")
  .option("kafka.security.protocol", "SASL_SSL")
  .option("kafka.sasl.mechanism", "PLAIN")
  .option("kafka.ssl.truststore.location", "/location)
  .option("kafka.ssl.truststore.password", "pass")
  .option("kafka.sasl.jaas.config", """jaasToUse""")
  .load()
  .writeStream
  .outputMode("append")
  .format("console")
  .option("startingOffsets", "earliest")
  .start().awaitTermination()

我有一个格式为的 Spark 数据集 -

+--------------+--------+-----+
|name          |type    |cost |
+--------------+--------+-----+
|AAAAAAAAAAAAAA|XXXXX   |0.24|
|AAAAAAAAAAAAAA|YYYYY   |1.14|
|BBBBBBBBBBBBBB|XXXXX   |0.78|
|BBBBBBBBBBBBBB|YYYYY   |2.67|
|BBBBBBBBBBBBBB|ZZZZZ   |0.15|
|CCCCCCCCCCCCCC|XXXXX   |1.86|
|CCCCCCCCCCCCCC|YYYYY   |1.50|
|CCCCCCCCCCCCCC|ZZZZZ   |1.00|
+--------------+--------+----+

我想将其转换为类型的对象 -

public class CostPerName {
    private String name;
    private Map<String, Double> costTypeMap;
}

我想要的是，

+--------------+-----------------------------------------------+
|name          |           typeCost.                           |
+--------------+-----------------------------------------------+
|AAAAAAAAAAAAAA|(XXXXX, 0.24), (YYYYY, 1.14)                   |            
|BBBBBBBBBBBBBB|(XXXXX, 0.78), (YYYYY, 2.67), (ZZZZZ, 0.15)    |
|CCCCCCCCCCCCCC|(XXXXX, 1.86), (YYYYY, 1.50), (ZZZZZ, 1.00)    |
+--------------+-----------------------------------------------+

即，对于每个name，我想要一张 的地图(type, cost)。

实现这一转变的有效途径是什么？我可以使用一些数据帧转换吗？我尝试了 groupBy 但只有在执行 sum、avg 等聚合查询时才有效。

我正在尝试使用 localhost:2181 的 zookeeper 访问在本地计算机上运行的 hbase。我安装了 phoenix-3.3.1-bin 并试图访问一个已经存在的 hbase tabe，但不能。因此，为了测试，我使用 phoenix 命令行创建了一个表，并在运行 !tables 命令时查看它。但是当我运行 selet 命令时，它显示错误。

这就是我正在做的，我使用的是 mac，hbase-0.94.26。同样的事情也发生在 squirrel-sql 客户端上。

0: jdbc:phoenix:localhost> CREATE TABLE stats.prod_metrics ( host char(50) not null, created_date date not null,
. . . . . . . . . . . . .>     txn_count bigint CONSTRAINT pk PRIMARY KEY (host, created_date) );
No rows affected (1.82 seconds)

0: jdbc:phoenix:localhost> !tables
+------------------------------------------+------------------------------------------+------------------------------------------+---------------------------+
|                TABLE_CAT                 |               TABLE_SCHEM                |                TABLE_NAME                |                TABLE_TYPE |
+------------------------------------------+------------------------------------------+------------------------------------------+---------------------------+
|                                          | SYSTEM                                   | CATALOG …