Sanzhar Sarsekov's questions

我在 Docker 容器中有一个数据库，我将它传输到服务器，但是在转储之后，我发现数据大小有 2-10 mb 的不同。只是没有发现数据丢失

为什么会发生这种情况

我像这样从docker转储数据

sudo docker exec -i docker_container pg_dump -U username -d database > /home/plazun/dump.sql

sudo -u postgres -i

psql -d database < /home/plazun/dump.sql

在码头工人

                           List of relations
 Schema |      Name      | Type  |   Owner    |   Size   | Description
--------+----------------+-------+------------+----------+-------------
 public | table1         | table | owner      | 24 MB    |
 public | table2         | table | owner      | 127 MB   |
 public | table3         | table | owner      | 16 kB    |
 public | table4         | table | owner      | 10232 kB |
 public | table5         | table | owner      | 320 kB   |
(5 rows)

在服务器上

                           List of relations
 Schema |      Name      | Type  |   Owner    |   Size   | Description
--------+----------------+-------+------------+----------+-------------
 public | table1         | table | owner      | 19 MB    |
 public | table2         | table | owner      | 71  MB   |
 public | table3         | table | owner      | 16 kB    |
 public | table4         | table | owner      | 10216 kB |
 public | table5         | table | owner      | 296 kB   |
(5 rows)

我是 Kubernetes 的新手。我注意到当我重新启动系统时，minikube 在 vi​​rtualbox 中停止。我怎样才能让它在系统重新启动时自动打开？

我正在尝试通过 psql 将 csv 文件中的数据解析到 postgresql，但我不明白如何编写查询以将数据从 csv 写入表，该表应返回一个 ID，我将立即将其写入另一个表多对一链接

这是写入一个表的示例

\copy table1(column1, column2...) from 'path' delimiter ';' csv header

仅使用 spark sql 时，对数据库的请求处理得非常快，但是当 JavaPairJDD 连接时，它开始变慢。

我想知道是否可以加快数据处理速度或以某种方式改进代码以使其不会冻结？

这是没有 JavaPairRDD 的运行

collectAsList at StackOverFlow.java:56, took 0.883860 s

在这里使用 JavaPairRDD

collectAsMap at StackOverFlow.java:61, took 128.340516 s

这是代码

public List<Order> getAllWithoutPairRDD(Optional<String> search, Optional<Integer> size) {

    SparkSession session = SparkSession.builder().config(config).getOrCreate();

    Properties properties = new Properties();

    properties.setProperty("partitionColumn", "id");
    properties.setProperty("fetchsize", "1000");
    properties.setProperty("driver", "org.postgresql.Driver");
    properties.setProperty("user", "postgres");
    properties.setProperty("password", "password");

    Dataset<Row> jdbc = session.read().jdbc("jdbc:postgresql://localhost/orders"
            , "orders"
            , "id"
            , 1L
            , 60000000L
            , 100
            , properties
    );
    Dataset<Row> ordersData = jdbc.select(col("*")).where(col("city_id").equalTo(3L)).limit(size.orElse(1));
    JavaPairRDD<Timestamp, Integer> analyticPairRDD = ordersData.toJavaRDD().mapToPair((PairFunction<Row, Timestamp, Integer>) row -> new Tuple2<Timestamp, Integer>((Timestamp) row.get(0), 1));

    JavaPairRDD<Timestamp,Integer> result = analyticPairRDD.groupByKey().mapValues(Iterables::size);
    return ordersData.as(orderEncoder).collectAsList();
}

public Map<Timestamp, Integer> getAllWithPairRDD(Optional<String> search, Optional<Integer> size) {

    SparkSession session = SparkSession.builder().config(config).getOrCreate();

    Properties properties = new Properties();

    properties.setProperty("partitionColumn", "id");
    properties.setProperty("fetchsize", "1000");
    properties.setProperty("driver", "org.postgresql.Driver");
    properties.setProperty("user", "postgres");
    properties.setProperty("password", "password");

    Dataset<Row> jdbc = session.read().jdbc("jdbc:postgresql://localhost/orders"
            , "orders"
            , "id"
            , 1L
            , 60000000L
            , 100
            , properties
    );
    Dataset<Row> ordersData = jdbc.select(col("operation_date")).where(col("city_id").equalTo(3L)).limit(size.orElse(1));
    JavaPairRDD<Timestamp, Integer> analyticPairRDD = ordersData.toJavaRDD().mapToPair((PairFunction<Row, Timestamp, Integer>) row -> new Tuple2<Timestamp, Integer>((Timestamp) row.get(0), 1));

    JavaPairRDD<Timestamp,Integer> result = analyticPairRDD.groupByKey().mapValues(Iterables::size);
    return result.collectAsMap();
}

@Bean
public SparkConf sparkConf() {
    return new SparkConf()
            .setAppName(appName)
            .setMaster(masterUri)
            .set("spark.executor.memory", "2g")
            .set("spark.executor.cores", "10")
            .set("spark.executor.memoryOverhead", "1g")
            .set("spark.driver.cores", "10")
            .set("spark.driver.memory", "3g")
            .set("spark.yarn.am.memory", "2g")
            .set("spark.yarn.am.cores", "4")
            .set("spark.sql.shuffle.partitions", "1000")
            .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
}

我正在使用带有 spark sql 的 hadoop。通常当我拉出 1000 列时，需要 1-2 秒，但是当超过 2000 列时，spark 会冻结并在几分钟内加载数据

spark-submit --deploy-mode client --master yarn --num-executors=40 --executor-cores=2 --executor-memory=5G  /home/hdoop/SparkApplication/demo-0.0.1-SNAPSHOT.jar 

@Configuration
@EnableAutoConfiguration(exclude = {org.springframework.boot.autoconfigure.gson.GsonAutoConfiguration.class})
public class SparkConfig {

    @Value("Test Spark Application")
    private String appName;
    @Value("local[*]")
    private String masterUri;

    @Bean
    public SparkConf sparkConf() {
        return new SparkConf()
                .setAppName(appName)
                .setMaster(masterUri)
                .set("spark.sql.debug.maxToStringFields", "1000")
                .setJars(new String[]
                {"/home/hdoop/SparkApplication/demo-0.0.1-SNAPSHOT.jar"
                ,"/home/hdoop/SparkApplication/spark-core_2.12-3.0.1.jar"
                ,"/home/hdoop/SparkApplication/postgresql-42.2.10.jar"});
    }

@Service
@EnableAutoConfiguration(exclude = {org.springframework.boot.autoconfigure.gson.GsonAutoConfiguration.class})
public class StackOverFlow implements Serializable {

    @Autowired
    private SparkConf sparkConf;

    public List<Order> getObject(String param, String value, Long limit) {
        Encoder<Order> orderEncoder = Encoders.bean(Order.class);

        SparkSession session = SparkSession
                .builder()
                .config(sparkConf)
                .getOrCreate();

        if (!session.sqlContext().isTraceEnabled()) {
            SparkSession.setActiveSession(session);
        }
                                                                        
        Dataset<Row> jdbcDF = session.read()
                .format("jdbc")
                .option("url", "jdbc:postgresql:postgres:5432//orders")
                .option("driver", "org.postgresql.Driver")
                .option("query", "select * from orders o where " + param + " = '" + value + "' limit " + limit)
                .option("user", "root")
                .option("password", "password")
                .load();

        List<Order> orders = jdbcDF.as(orderEncoder).collectAsList();
        session.stop();
        return orders;
    }

如何转换Data<Row> в javaRDD<User>？

我试图从数据库中提取一个包含大量列的列表并遇到这样一个问题，我必须指定每列的位置及其值，如果没有这一切，我怎么能得到它？

这是我的代码

        SparkConf sparkConf = new SparkConf()
            .setMaster("local[*]")
            .setAppName("test");

    SparkContext sparkContext = new SparkContext(sparkConf);

    SparkSession session = new SparkSession(sparkContext);

    DataFrameReader dataFrameReader = session
            .read()
            .format("jdbc")
            .option("url", "jdbc:postgresql://10.100.0.4:5432/refund_service")
            .option("driver", "org.postgresql.Driver")
            .option("dbtable", "refunds")
            .option("user", "smartplaza")
            .option("password", "smartplaza");

    Encoder<Something> somethingEncoder = Encoders.bean(Refund.class);

    Dataset<Row> response = dataFrameReader.load();

    JavaRDD<Something> rsomethingJavaRDD = response.javaRDD().map(new Function<Row, Refund>() {
        @Override
        public Something call(Row row) throws Exception {
            return new Something(row.getLong(0),
                    row.getTimestamp(1), row.getTimestamp(2),row.getTimestamp(3),
                    row.getDouble(4),
                    row.getDouble(5),
                    row.getDouble(6),
                    row.getLong(7),
                    row.getLong(8),
                    row.getDouble(9),
                    row.getLong(10),
                    row.getDouble(11),
                    (Long) row.get(12),
                    (Long) row.get(13),
                    (Long) row.get(14),
                    row.getBoolean(15),
                    (Long) row.get(16),
                    (Long) row.get(17),
                    row.getDouble(18),
                    row.getDouble(19),
                    (Long) row.get(20),
                    (Long)row.get(21));
        }
    });

    Dataset<Something> somethingDataset = session.createDataset(refundJavaRDD.rdd(),somethingEncoder);