如果我事先不知道使用Apache Flink的模式，有没有办法将数据写入到拼图文件中？
是的，即使在不了解Apache Flink的情况下，也有办法将数据写入到拼图文件中。
拼图文件是一种用于存储大规模数据集的文件格式，它具有高效的压缩和读取性能。在云计算领域，拼图文件常用于大数据处理和分析任务。
要将数据写入到拼图文件中，可以使用Apache Parquet库。Apache Parquet是一种列式存储格式，它能够高效地存储和处理大规模数据集。
在使用Apache Flink时，可以通过以下步骤将数据写入到拼图文件中：
- 导入所需的依赖库：
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
import import org.apache.flink.core.fs.Path;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.formats.parquet.avro.ParquetAvroWriters;- 创建Flink的执行环境：
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();- 准备要写入的数据集：
DataSet<Tuple2<String, Integer>> data = ...- 将数据写入到拼图文件中：
data.writeAsFormattedText("hdfs://path/to/parquet/file", FileSystem.WriteMode.OVERWRITE)
.setParallelism(1)
.name("Write to Parquet")
.writeUsingOutputFormat(ParquetAvroWriters.forReflectRecord(data.get(0).getClass()))
.setParallelism(1)
.name("Parquet Writer");在上述代码中，data是要写入的数据集，可以根据实际情况进行替换。"hdfs://path/to/parquet/file"是拼图文件的路径，可以根据实际需求进行修改。
需要注意的是，上述代码中使用了Hadoop分布式文件系统（HDFS）作为拼图文件的存储介质。如果要将数据写入到其他存储介质，可以相应地修改路径。
推荐的腾讯云相关产品是腾讯云对象存储（COS），它是一种高可用、高可靠、低成本的云存储服务。您可以将拼图文件存储在腾讯云COS中，以实现数据的持久化存储和高效读取。
腾讯云COS产品介绍链接地址：https://cloud.tencent.com/product/cos
相关·内容
我想写一些数据集给蜂巢。我试过使用hive，但它不支持batchExecute。因此，我修改了它，将其写入hdfs，然后生成hive表。
我尝试使用以下代码编写hdfs：
package test;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.hadoop.
浏览 0提问于2018-10-31得票数 0
回答已采纳
在flink项目中，我使用case类单击。
case class click( date: LocalDateTime, stbId:String, channelId :Int)
这个类填充了DataSets，并且在日期为java 8 java.time.LocalDateTime的情况下工作得很好。在java7环境中切换到org.joda (版本2.9)之后，对DataSets中的click对象的调用没有像以前那样执行。访问单击对象的日期字段的某些函数会引发NullPointerExceptions。这些函数的例子是getHourOfDay、toString等。我能够确保click类的da
浏览 11提问于2015-11-11得票数 4
回答已采纳
我在上读到了用于单词计数的Flink批处理模式
在我看来：
对于批处理，我需要ExecutionEnvironment，就像val env = ExecutionEnvironment.getExecutionEnvironment
对于流，我需要StreamExecutionEnvironmetn，就像val env = StreamExecutionEnvironment.getExecutionEnvironment
我一直对这些执行环境感到困惑。由于Flink被认为是统一的批处理和处理框架，我认为执行环境的创建应该仅仅是one，例如：
val env = Executi
浏览 3提问于2021-05-29得票数 0
回答已采纳
我尝试做以下几件事：
env
.readHadoopFile(new TeraInputFormat(), classOf[Text], classOf[Text], inputPath)
.map(tp => tp)
但是在我的编辑器中出现了一个类型不匹配的错误：
Expected: MapFunction[Tuple2[Text, Text], NotInferedR], actual: (Nothing) => Nothing
我该如何解决这个问题呢？
这是完整的代码：
import org.apache.flink.api.common.functions.Part
浏览 10提问于2016-07-07得票数 2
回答已采纳
我们想要使用s3中的拼图文件
我的代码片段是这样的。我的输入文件是protobuf编码的拼图文件。protobuf类是Pageview.class。
import com.twitter.chill.protobuf.ProtobufSerializer;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.scala.hadoop.mapreduce.HadoopInpu
浏览 6提问于2019-01-21得票数 0
我试图通过使用BucketingSink和自定义ParquetSinkWriter在HDFS上使用Apache编写拼花文件。
下面是代码和上面的错误，指出当启用检查点(在BucketingSink类中调用BucketingSink())下的刷新方法不安静工作时。即使是编写器都以“writer.close()”结尾，但仍然从"writer = createWriter();“中获得错误。有什么想法吗？谢谢
犯了这样的错误：
客户端的org.apache.hadoop.fs.FileAlreadyExistsException: /user/hive/flink_parquet_fils_
浏览 1提问于2020-02-07得票数 1
2回答
我试图以本地模式在Apache中执行一个示例程序。
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
public class WordCountExam
浏览 2提问于2015-06-01得票数 7
我是flink的新手(就这一点而言，还包括parquet/hadoop )，所以我肯定是在做一些非常愚蠢的事情。我正在尝试创建一个接收器，它会将我的数据源转储到一个拼图文件中。
我的代码看起来像这样：
val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
streamEnv.setParallelism(1);
streamEnv.enableCheckpointing(100, CheckpointingMode.EXACTLY_ONCE);
val sink = StreamingFileSink.forBu
浏览 0提问于2018-10-04得票数 2
1回答
我的拼图文件的当前模式是：
scala> df.schema
org.apache.spark.sql.types.StructType = StructField(route,StringType,true), StructField(wgt,StringType,true))
我想将其更改为以下内容，并将其保存并覆盖到我的HDFS中：
org.apache.spark.sql.types.StructType = StructField(route,StringType,true), StructField(wgt,DoubleType,true))
将模式转换为DoubleType
浏览 11提问于2021-06-02得票数 0
我试图用Flink编写一个计算，它需要两个阶段。
在第一个阶段，我创建一个图并获取它的顶点ids：
List<String> ids = graph.getVertexIds().collect();
在第二阶段，我希望使用这些ids来运行每个顶点的SingleSourceShortestPath。
for (String id: ids){
System.out.println("Source Id: "+id);
graph.run( new SingleSourceShortestPaths<String, String&
浏览 6提问于2017-11-28得票数 0
回答已采纳
我已经为BucketingSink创建了一个编写器。接收器和写入器可以正常工作，但当写入器将avro genericrecord写入到拼图时，该文件是从正在进行中创建的，等待完成。但是这些文件是0字节的空文件。有人能告诉我代码出了什么问题吗？我尝试将AvroParquetWriter的初始化放在open()方法中，但结果仍然相同。
在调试代码时，我确认writer.write( element )确实已执行，并且element包含avro genericrecord数据
流式数据
BucketingSink<DataEventRecord> sink =
new Bucke
浏览 6提问于2017-02-17得票数 3
我正在为集成hdfs到flink而苦苦挣扎。
Scala二进制版本: 2.12，
Flink (集群)版本: 1.10.1
这个配置和HADOOP_CONF_DIR在任务管理器中也是一样的。
pom.xml；
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId&
浏览 2提问于2020-06-24得票数 0
2回答
我有一个用例，当我们在模式注册表中为主题起诉TopicRecordNameStrategy时，在单个Kafka主题中有多种类型的Avro记录。
现在，我已经编写了一个使用者来阅读该主题并构建一个GenericRecord数据流。现在，我不能将此流汇到parquet格式的hdfs/s3中，因为此流包含不同类型的模式记录。因此，我通过应用过滤器并创建不同的流，然后分别下沉每个流，为每种类型过滤不同的记录。
下面是我使用的代码-
import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
import io
浏览 69提问于2020-02-21得票数 1
回答已采纳
1回答
如何在进行bazel构建时传递参数。实际上，我的程序会接受两个参数，一个是目录，另一个是target.csv，我需要把目录下的文件写到csv中。
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collector;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apac
浏览 7提问于2017-04-11得票数 1
1回答
在当前的环境中，我有一个加密的hdfs区域。我使用Flink 1.2.0集群将parquet文件写入hdfs加密区域，使用Apache编写器版本1.9.0。我在写入非加密区域时没有问题，但是当我写信到加密区域时，我有以下错误：
org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.UnknownCryptoProtocolVersionException): No crypto protocol versions provided by the client are supported. Client provided: []
浏览 5提问于2017-04-03得票数 1
回答已采纳
我有一个调度器，它可以获取集群指标，并使用旧版本的Cloudera API将数据写入HDFS文件。但最近，我们更新了JAR和原始代码错误，但出现了一个异常。
java.lang.ClassCastException: org.apache.hadoop.io.ArrayWritable cannot be cast to org.apache.hadoop.hive.serde2.io.ParquetHiveRecord
at org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriteSupport.write(DataWritab
浏览 1提问于2017-04-13得票数 1
5回答
我正在尝试将拼图文件作为数据帧读取，它将定期更新(路径为/folder_name。每当有新数据到来时，旧的地块文件路径(/folder_name)将被重命名为临时路径，然后我们合并新数据和旧数据并将其存储在旧路径(/folder_name)中。
发生的情况是，假设我们在更新之前有一个拼图文件作为hdfs://folder_name/part-xxxx-xxx.snappy.parquet，然后在更新之后将其更改为hdfs://folder_name/part-00000-yyyy-yyy.snappy.parquet
当我尝试在更新过程中读取拼图文件时，会出现问题
sparksession.r
浏览 1提问于2019-11-08得票数 4
1回答
我有以下flink work count，当我在IDE中运行它时，它会正确地打印字数，如下所示 (hi,2)
(are,1)
(you,1)
(how,1) 但是当我在集群中运行它时，我没有找到输出。 1. Start cluster using start-cluster.sh
2. Open the webui at http://localhost:8081
3. In the Submit new Job page, Submit the jar, and then input the entry class and then click the Submit button to s
浏览 16提问于2019-09-14得票数 0
我正在IntelliJ中创建和测试一个机器学习模型，该模型将对来自流的传入数据进行分类。我正在Scala (2.11.8)和使用Flink框架(1.8.3)中工作。我正在尝试读取一个csv文件，该文件包含具有长、双和布尔数据类型的列中的培训数据，这些数据类型如下：
val env = ExecutionEnvironment.getExecutionEnvironment
val trainingDS = env.readCsvFile(./training.csv)(DataSet[(Long, Double, Boolean)])
我得到的错误是：“未指定的值参数证据$2”。此参数对应于T
浏览 10提问于2020-08-12得票数 1
1回答
我有一个由分区的拼图文件组成的大型Impala数据库。我直接使用HDFS将一个Parquet分区复制到本地磁盘。此分区总共有15 of，由许多文件组成，每个文件大小为10MB。我正在尝试使用Pandas与Pyarrow引擎或Pyarrow直接读取，但它在内存中的大小使用了超过60‘t的RAM，并且它在使用所有内存之前不会读取整个数据集。内存使用量如此之大的原因可能是什么？
浏览 32提问于2019-02-02得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
