基本上,我有一个火花流工作(与增量)写一个小文件到hdfs每5分钟。我还有一个压缩作业,每天运行,将前一天的数据压缩到一些大文件中(文件的#取决于作业重新分区号)。大文件与原始小文件位于同一个目录中。是否有任何方法有效地删除原来的小文件,因为它们是无用的?
我已经尝试过用于增量表的vacuum函数,但这基本上将所有数据从保留期中删除,不管是否压缩。
下面是我如何压缩数据(我正在使用Java):
spark.read()
.format("delta")
.load(path) // hdfs path of the data
.w
我使用下面的查询创建了一个单元表,并使用下面提到的第二个查询,每天将数据插入到这个表中。
create EXTERNAL table IF NOT EXISTS DB.efficacy
(
product string,
TP_Silent INT,
TP_Active INT,
server_date date
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
STORED AS TEXTFILE
LOCATION 'hdfs://hdfsadlproduction/user/DB/Report/efficacy'
我知道这听起来很愚蠢,也明白hadoop不是为小文件设计的,但不幸的是,我收到了6000+小文件,每个文件大小约为50kb。
每次我尝试运行"hadoop fs -put -f /path/FOLDER_ with _FILES /target/HDSF_FOLDER“时,当连接到namenode时,总会因为一个随机文件而失败。
java.net.SocketTimeoutException: 75000 millis timeout while waiting for channel to be ready for read. ch : java.nio.channels.Socke
我运行Spark查询并使用它们执行数据转换,然后将最终的结果集(在一系列转换步骤之后)存储到S3。
我最近注意到,我的工作之一是在编写S3时创建大量分区文件,并且需要很长时间才能完成(实际上是失败的)。因此,我想知道是否有任何方法在SQL中执行类似COALESCE的函数来在写入S3之前减少分区的数量?
我知道SQL等效于重新分区是Cluster By。因此,我想知道SQL中是否也有类似的COALESCE操作。
请注意,我只能访问SQL,所以我的问题仅限于Spark SQL API。(例如SELECT col from TABLE1 WHERE .)
我们使用SparkSQL2.4.6.7
谢谢
我有一个文本文件(大约600MB),其中包含使用搜索api收集的json格式的推文。我想根据tweet中的"created_at“值对tweet进行排序,并将它们写入.I使用代码的新文件中:
oldlist=[]
newlist=[]
with open('abc.txt') as f:
for each in f:
oldlist.append(json.loads(each))
newlist=(sorted(oldlist,key=lambda k: k['created_at'])).
我可以使用“>>”将输出
我已经尝试了几个选项,但我只看到配置设置将小文件合并为大文件,如下所示,但反之亦然。我希望创建大小为150kb的文件。
set hive.merge.tezfiles=true;
set hive.merge.smallfiles.avgsize=128000;
set hive.merge.size.per.task=128000;
我正在尝试通过SparkSQL选择周日录制的所有记录。我做了以下几次尝试,但都是徒劳的。
SELECT * FROM mytable WHERE DATEPART(WEEKDAY, create_time) = 0
SELECT * FROM mytable WHERE strftime("%w", create_time) = 0
如何在SparkSQL中获取星期几?
我需要使用sparksql.I在我的hive表中插入一些值。我使用以下代码。 val filepath:String = "/user/usename/filename.csv'"
val fileName : String = filepath
val result = fileName.split("/")
val fn=result(3) //filename
val e=LocalDateTime.now() //timestamp 首先,我尝试使用Insert Into Values,但后来我发现此
我已经创建了一个Azure data Lake gen2文件系统来存储和恢复数据。我已经正确地加载了我的数据(每天一个文件),但是,当我想要使用Azure Data Lake Azure Rest API获取数据时,我只能针对每个请求访问一个文件,所以,如果我需要一个月,我必须每天执行一个请求并将其合并到我的本地机器中,而直接在Azure Data Lake gen2中这样做是可取的。
我知道这可以在数据库中使用SparkSQL和通配符来完成,但是如果我可以直接在Azure Data Lake gen2 rest API中使用通配符,那就更好了。我已经找过了,但是我发现文档不清楚。每个人都知道