标签: stream

我想从服务器读取 xls。我成功地使用 WebClient 下载了文件，但现在我想在不保存文件的情况下打开它。

        var request = WebRequest.Create(location);
        var response = request.GetResponse();
        var stream = response.GetResponseStream();
        var document = SpreadsheetDocument.Open(stream, false);
        if (stream != null) stream.Close();

我的代码在 SpreadsheetDocument.Open 上停止，它给出了以下错误：

无法对不支持查找的流进行操作。

我做错了什么，为什么我无法打开文档？

我正在尝试使用节点请求模块从第三方服务获取一些数据，并将此数据作为字符串从函数返回。我的看法是request()返回一个可读的流，因为你可以这样做request(...).pipe(writeableStream)- 我认为 - 意味着我可以做到

function getData(){
    var string;

    request('someurl')
        .on('data', function(data){
             string += data;
        })
        .on('end', function(){
            return string;
        });
}

但这并不奏效。我想我对 request() 或节点流的真正工作方式有一些错误的看法。有人可以在这里澄清我的困惑吗？

我使用项目的 Apache Tika 包依赖项来找出文件的 MimeTypes。由于一些问题，我们必须通过 InputStream 来查找。它实际上保证标记/重置给定的 InputStream。Tika-Bundle 包括核心和解析器 api，并使用 PoifscontainerDetector、ZipContainerDetector、OggDetector、MimeTypes 和 Magic 进行检测。我已经调试了 3 个小时，所有的 Detectors 都在检测后标记并重置。我是按照以下方式做的。

  TikaInputStream tis = null;
    try {
        TikaConfig config = new TikaConfig();
        tikaDetector = config.getDetector();
        tis =  TikaInputStream.get(in);
        MediaType mediaType = tikaDetector.detect(tis, new Metadata());

        if (mediaType != null) {
            String[] types = mediaType.toString().split(",");

            for (int i = 0; i < types.length; i++) {
                mimeTypes.add(new MimeType(types[i]));
            }
        }

    } catch (Exception e) {
        logger.error("Mime Type for given Stream could not be resolved: ", e); …

我有一个流，其下N个字节是 UTF8 编码的字符串。我想以最少的开销创建该字符串。

这有效：

var bytes = new byte[n];
stream.Read(bytes, 0, n); // my actual code checks return value
var str = Encoding.UTF8.GetString(bytes);

在我的基准测试中，我看到花费了大量时间以byte[]临时形式收集垃圾。如果我可以摆脱这些，我可以有效地将我的堆分配减半。

本UTF8Encoding类没有与流工作方法。

如果有帮助，我可以使用不安全的代码。我不能重用一个byte[]缓冲区，如果没有ThreadLocal<byte[]>它似乎引入了比它减轻的更多的开销。我确实需要支持 UTF8（ASCII 不会削减它）。

这里有我缺少的 API 或技术吗？

通过 抓取图像GetStreamAsync，如何确定状态？

HttpClient OpenClient = new HttpClient();
Stream firstImageStream = OpenClient.GetStreamAsync("imageUrl.jpg").Result;

有时这会产生错误（通常为 403 或 404），我只想跳过处理这些结果。

我能找到的所有内容都说要使用StatusCode属性 or IsSuccessStatusCode，但那些似乎只适用于HttpResponseMessage来自 GetAsync 的 ，它没有给我Stream处理图像所需的条件。

我正在编写一个组件来从 HTTP 服务流式传输一个大的（4+ GB）。该组件采用 URL 和目标流。目标流可以是文件流，也可以是 POST 到不同 HTTP 服务的流，甚至两者都可以。作为组件的作者，我需要执行以下步骤，直到完成：

1. 从 HTTP 流中读取一个合理大小的缓冲区，
2. 将此缓冲区写入目标流，
3. 刷新目标流（输出到磁盘、网络等）

我永远不应该有超过内存中数据缓冲区的大小。

我正在使用 flurl 对服务器进行 HTTP 调用。我已经尝试了以下方法来拨打我的电话

var stream = await flurlClient.GetStreamAsync();

这给了我一个 MemoryStream，它不起作用，因为它会填满并占用与文件大小一样多的内存。

var response = flurlClient.GetAsync();
var stream = response.Content.ReadAsStreamAsync();

再次，内存流。

var response = flurlClient.GetAsync();
var stream = new CustomFlushingSteam();
response.Content.CopyToAsync(stream);

这个看起来很有希望，但遗憾的是，它试图使用单个 Write() 语句编写整个内容。

我怎样才能在不破坏记忆的情况下完成这项任务？我更喜欢使用 flurl，但我并没有被它束缚。

我正在尝试使用这些典型代码来积累推文数据。如您所见，我尝试跟踪包含“UniversalStudios”、“Disneyland”或“Los Angeles”的推文。但事实上，我真正想要的是包含这些关键词“UniversalStudios”、“迪士尼乐园”和“洛杉矶”的推文。谁能告诉我如何实现这一目标？

非常感谢提前:)

#This is a basic listener that just prints received tweets to stdout.
class StdOutListener(StreamListener):

    def on_data(self, data):
        all_data = json.loads(data)
        tweet = TextBlob(all_data["text"])

        #Add the 'sentiment data to all_data
        #all_data['sentiment'] = tweet.sentiment

        #print(tweet)
        #print(tweet.sentiment)

        # Open json text file to save the tweets
        with open('tweets.json', 'a') as tf:
            # Write a new line
            tf.write('\n')

            # Write the json data directly to the file
            json.dump(all_data, tf)
            # Alternatively: tf.write(json.dumps(all_data))
        return True

    def on_error(self, status):
        print (status)


if __name__ …

我想知道是否可以通过使用Flink的摄取时间模式来获取记录的时间戳。考虑以下 flink 代码示例（https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples /join/WindowJoinSampleData.scala ),

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)

val grades = WindowJoinSampleData.getGradeSource(env, rate)
val salaries = WindowJoinSampleData.getSalarySource(env, rate)

val joined = joinStreams(grades, salaries, windowSize)

...
case class Grade(name: String, level: Int) 
case class Salary(name: String, salary: Int)

默认情况下，等级和工资都不包含时间戳字段。但是，由于 Flink 允许使用“ingestionTime”为数据流中的记录分配挂钟时间戳，是否有可能在运行时获得这样的时间戳？例如，这是我想要做的：

val oldDatastream = env.addSource...  // Using ingestion time
val newDatastream = oldDatastream.map{record =>   
    val ts = getRecordTimestamp(record)
    // do some thing with ts
    }

谢谢你的帮助。

我开始使用 KStream 来使用来自现有主题的数据。

我只对在 10 秒窗口内获取给定 ID 的最后一个事件感兴趣。我尝试使用以下代码：

StreamsBuilder builder = new StreamsBuilder();
KStream<String, MySale> stream = builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), specificAvroSerde));

stream.selectKey((key, value) -> value.getID())
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
    .reduce((value1, value2) -> value2)
    .toStream()
    .peek((key, value) -> log.info("key={}, value={}", key, value.toString()))
    .to(OUTPUT_TOPIC, Produced.with(Serdes.String(), specificAvroSerde));

但我最终得到了所有事件，而不仅仅是最后一个。使用 KStream 可以做我想做的事吗？

使用 React/Express 处理 SSR 应用程序时，我试图掌握 renderToNodeStream 和一般流。

我有一个大页面（400kb 未压缩）并且使用 renderToNodeStream 提供了一个非常好的 TTFB（到第一个字节的时间），我想念的是一些压缩以使发送回的 HTML 响应更小但我不能让它与 renderToNodeStream 一起工作。

我错过了什么吗？
是否可以流式传输新鲜呈现的响应并压缩它们？

const stream = renderToNodeStream(<MyApp/>)

// this doesn't work
stream.pipe(zlib.createGzip())
stream._flush = zlib.Z_SYNC_FLUSH

stream.pipe(
 res,
 { end: "false" }
)

// stream.on("data", data => {
//   console.log(data)
// })
stream.on("end", () => {
  res.end(pageEnd())
})